From efc87242b2e295bc127b3a38e4039eaa0e83ea5c Mon Sep 17 00:00:00 2001 From: Vijay Pandian Date: Thu, 30 Jul 2026 23:34:27 -0400 Subject: [PATCH 1/3] [orchagent]: Option 3: in-process SAI notification queue for ZMQ mode Add shared SaiNotificationQueue, dispatcher, and SaiNotificationOrch. Migrate SAI notification handlers to enqueue in ZMQ mode and dispatch via handleNotification() across Fdb, Ports, Bfd, Icmp, Twamp, Dash, MACsec, and HFTel orchs. Signed-off-by: Vijay Pandian --- orchagent/Makefile.am | 1 + orchagent/bfdorch.cpp | 68 +- orchagent/bfdorch.h | 2 + orchagent/dash/dashhafloworch.cpp | 26 +- orchagent/dash/dashhafloworch.h | 1 + orchagent/dash/dashhaorch.cpp | 230 +++--- orchagent/dash/dashhaorch.h | 3 + orchagent/fdborch.cpp | 67 +- orchagent/fdborch.h | 2 + .../high_frequency_telemetry/hftelorch.cpp | 31 +- .../high_frequency_telemetry/hftelorch.h | 2 + orchagent/icmporch.cpp | 89 ++- orchagent/icmporch.h | 2 + orchagent/macsecorch.cpp | 38 +- orchagent/macsecorch.h | 3 +- orchagent/notifications.cpp | 327 ++++++-- orchagent/notifications.h | 64 ++ orchagent/orchdaemon.cpp | 13 +- orchagent/portsorch.cpp | 167 ++-- orchagent/portsorch.h | 4 +- orchagent/sainotificationorch.cpp | 24 + orchagent/sainotificationorch.h | 38 + orchagent/twamporch.cpp | 111 +-- orchagent/twamporch.h | 2 + tests/mock_tests/Makefile.am | 1 + tests/mock_tests/notifications_ut.cpp | 727 +++++++++++++----- tests/mock_tests/portsorch_ut.cpp | 7 - 27 files changed, 1517 insertions(+), 533 deletions(-) create mode 100644 orchagent/sainotificationorch.cpp create mode 100644 orchagent/sainotificationorch.h diff --git a/orchagent/Makefile.am b/orchagent/Makefile.am index f535971c547..b1491f59938 100644 --- a/orchagent/Makefile.am +++ b/orchagent/Makefile.am @@ -56,6 +56,7 @@ orchagent_SOURCES = \ orchdaemon.cpp \ orch.cpp \ notifications.cpp \ + sainotificationorch.cpp \ nhgorch.cpp \ nhgbase.cpp \ cbf/cbfnhgorch.cpp \ diff --git a/orchagent/bfdorch.cpp b/orchagent/bfdorch.cpp index 57c0a92fd49..884fad41e46 100644 --- a/orchagent/bfdorch.cpp +++ b/orchagent/bfdorch.cpp @@ -8,6 +8,7 @@ #include "sai_serialize.h" #include "directory.h" #include "notifications.h" +#include "sainotificationorch.h" #include "schema.h" using namespace std; @@ -90,6 +91,16 @@ BfdOrch::BfdOrch(DBConnector *db, string tableName, TableConnector stateDbBfdSes } Orch::addExecutor(bfdStateNotificatier); register_state_change_notif = false; + + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_BFD_SESSION_STATE_CHANGE, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + } } BfdOrch::~BfdOrch(void) @@ -237,39 +248,50 @@ void BfdOrch::doTask(NotificationConsumer &consumer) return; } - if (op == "bfd_session_state_change") + KeyOpFieldsValuesTuple entry = std::make_tuple(data, op, values); + handleNotification(entry); +} + +void BfdOrch::handleNotification(KeyOpFieldsValuesTuple &entry) +{ + if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_BFD_SESSION_STATE_CHANGE) { - uint32_t count; - sai_bfd_session_state_notification_t *bfdSessionState = nullptr; + handleBfdSessionStateChangeNotification(kfvKey(entry)); + } +} - sai_deserialize_bfd_session_state_ntf(data, count, &bfdSessionState); +void BfdOrch::handleBfdSessionStateChangeNotification(const std::string &data) +{ + uint32_t count; + sai_bfd_session_state_notification_t *bfdSessionState = nullptr; - for (uint32_t i = 0; i < count; i++) - { - sai_object_id_t id = bfdSessionState[i].bfd_session_id; - sai_bfd_session_state_t state = bfdSessionState[i].session_state; + sai_deserialize_bfd_session_state_ntf(data, count, &bfdSessionState); - SWSS_LOG_INFO("Get BFD session state change notification id:%" PRIx64 " state: %s", id, session_state_lookup.at(state).c_str()); + for (uint32_t i = 0; i < count; i++) + { + sai_object_id_t id = bfdSessionState[i].bfd_session_id; + sai_bfd_session_state_t state = bfdSessionState[i].session_state; - if (state != bfd_session_lookup[id].state) - { - auto key = bfd_session_lookup[id].peer; - m_stateBfdSessionTable.hset(key, "state", session_state_lookup.at(state)); + SWSS_LOG_INFO("Get BFD session state change notification id:%" PRIx64 " state: %s", id, session_state_lookup.at(state).c_str()); - SWSS_LOG_NOTICE("BFD session state for %s changed from %s to %s", key.c_str(), - session_state_lookup.at(bfd_session_lookup[id].state).c_str(), session_state_lookup.at(state).c_str()); + if (state != bfd_session_lookup[id].state) + { + auto key = bfd_session_lookup[id].peer; + m_stateBfdSessionTable.hset(key, "state", session_state_lookup.at(state)); - BfdUpdate update; - update.peer = key; - update.state = state; - notify(SUBJECT_TYPE_BFD_SESSION_STATE_CHANGE, static_cast(&update)); + SWSS_LOG_NOTICE("BFD session state for %s changed from %s to %s", key.c_str(), + session_state_lookup.at(bfd_session_lookup[id].state).c_str(), session_state_lookup.at(state).c_str()); - bfd_session_lookup[id].state = state; - } - } + BfdUpdate update; + update.peer = key; + update.state = state; + notify(SUBJECT_TYPE_BFD_SESSION_STATE_CHANGE, static_cast(&update)); - sai_deserialize_free_bfd_session_state_ntf(count, bfdSessionState); + bfd_session_lookup[id].state = state; + } } + + sai_deserialize_free_bfd_session_state_ntf(count, bfdSessionState); } bool BfdOrch::register_bfd_state_change_notification(void) diff --git a/orchagent/bfdorch.h b/orchagent/bfdorch.h index 291abc9201c..04c1f5728cc 100644 --- a/orchagent/bfdorch.h +++ b/orchagent/bfdorch.h @@ -36,6 +36,8 @@ class BfdOrch: public Orch, public Subject uint32_t bfd_src_port(void); void notify_session_state_down(const std::string& key); + void handleNotification(swss::KeyOpFieldsValuesTuple &entry); + void handleBfdSessionStateChangeNotification(const std::string &data); bool register_bfd_state_change_notification(void); void update_port_number(std::vector &attrs); sai_status_t retry_create_bfd_session(sai_object_id_t &bfd_session_id, vector attrs); diff --git a/orchagent/dash/dashhafloworch.cpp b/orchagent/dash/dashhafloworch.cpp index 9108a4dbb52..2f60ebb1fb8 100644 --- a/orchagent/dash/dashhafloworch.cpp +++ b/orchagent/dash/dashhafloworch.cpp @@ -11,6 +11,7 @@ #include "macaddress.h" #include "swssnet.h" #include "schema.h" +#include "sainotificationorch.h" #include #include @@ -786,6 +787,16 @@ DashHaFlowOrch::DashHaFlowOrch(DBConnector *db, const vector &tableNames Orch::addExecutor(flowBulkGetSessionNotifier); registerFlowBulkGetSessionNotifier(); + + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + } } bool DashHaFlowOrch::registerFlowBulkGetSessionNotifier() @@ -854,13 +865,22 @@ void DashHaFlowOrch::doTask(NotificationConsumer &consumer) consumer.pop(notification_name, data, values); - if (notification_name == SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT) + KeyOpFieldsValuesTuple entry = std::make_tuple(data, notification_name, values); + handleNotification(entry); +} + +void DashHaFlowOrch::handleNotification(KeyOpFieldsValuesTuple &entry) +{ + if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT) { - handleSessionNotification(notification_name, data, values); + handleSessionNotification( + kfvOp(entry), + kfvKey(entry), + kfvFieldsValues(entry)); } else { - SWSS_LOG_WARN("Unknown notification: %s", notification_name.c_str()); + SWSS_LOG_WARN("Unknown notification: %s", kfvOp(entry).c_str()); } } diff --git a/orchagent/dash/dashhafloworch.h b/orchagent/dash/dashhafloworch.h index 8bbd0d07ef7..379885c514e 100644 --- a/orchagent/dash/dashhafloworch.h +++ b/orchagent/dash/dashhafloworch.h @@ -163,6 +163,7 @@ class DashHaFlowOrch : public ZmqOrch void doTask(ConsumerBase &consumer); void doTask(swss::NotificationConsumer &consumer); + void handleNotification(swss::KeyOpFieldsValuesTuple &entry); void doTask(swss::SelectableTimer &timer); void doTaskFlowSyncSessionTable(ConsumerBase &consumer); void doTaskFlowDumpFilterTable(ConsumerBase &consumer); diff --git a/orchagent/dash/dashhaorch.cpp b/orchagent/dash/dashhaorch.cpp index 7fdd63c1d79..32a19d8164f 100644 --- a/orchagent/dash/dashhaorch.cpp +++ b/orchagent/dash/dashhaorch.cpp @@ -12,6 +12,7 @@ #include "taskworker.h" #include "pbutils.h" #include "converter.h" +#include "sainotificationorch.h" #include "chrono" @@ -92,6 +93,22 @@ DashHaOrch::DashHaOrch(DBConnector *db, const vector &tables, DashOrch * register_ha_set_notifier(); register_ha_scope_notifier(); + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_HA_SET_EVENT, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_HA_SCOPE_EVENT, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + } + // Register this DashHaOrch instance with DashOrch m_dash_orch->setDashHaOrch(this); } @@ -1129,130 +1146,147 @@ void DashHaOrch::doTask(NotificationConsumer &consumer) for (auto &event : events) { - std::string op = kfvOp(event); - std::string data = kfvKey(event); - std::vector values = kfvFieldsValues(event); + handleNotification(event); + } +} - if (op == "ha_set_event") - { - std::time_t now_time = getNowTime(); +void DashHaOrch::handleNotification(KeyOpFieldsValuesTuple &entry) +{ + const auto &op = kfvOp(entry); + const auto &data = kfvKey(entry); + + if (op == SAI_SWITCH_NOTIFICATION_NAME_HA_SET_EVENT) + { + handleHaSetEventNotification(data); + } + else if (op == SAI_SWITCH_NOTIFICATION_NAME_HA_SCOPE_EVENT) + { + handleHaScopeEventNotification(data); + } +} - uint32_t count; - sai_ha_set_event_data_t *ha_set_event = nullptr; +void DashHaOrch::handleHaSetEventNotification(const std::string &data) +{ + SWSS_LOG_ENTER(); - sai_deserialize_ha_set_event_ntf(data, count, &ha_set_event); + std::time_t now_time = getNowTime(); - for (uint32_t i = 0; i < count; i++) - { - sai_object_id_t ha_set_id = ha_set_event[i].ha_set_id; - sai_ha_set_event_t event_type = ha_set_event[i].event_type; + uint32_t count; + sai_ha_set_event_data_t *ha_set_event = nullptr; - SWSS_LOG_INFO("Get HA Set event notification id:%" PRIx64 " event: Data plane channel goes %s", ha_set_id, sai_ha_set_event_type_name.at(event_type).c_str()); + sai_deserialize_ha_set_event_ntf(data, count, &ha_set_event); - auto key = getHaSetObjectKey(ha_set_id); - if (key.empty()) - { - SWSS_LOG_ERROR("HA Set object not found for ID: %" PRIx64, ha_set_id); - continue; - } - std::vector fvs = { - {"last_updated_time", to_string(now_time)}, - {"dp_channel_is_alive", sai_ha_set_event_type_name.at(event_type)} - }; - m_dpuStateDbHaSetTable->set(key, fvs); - } - sai_deserialize_free_ha_set_event_ntf(count, ha_set_event); - } + for (uint32_t i = 0; i < count; i++) + { + sai_object_id_t ha_set_id = ha_set_event[i].ha_set_id; + sai_ha_set_event_t event_type = ha_set_event[i].event_type; + + SWSS_LOG_INFO("Get HA Set event notification id:%" PRIx64 " event: Data plane channel goes %s", ha_set_id, sai_ha_set_event_type_name.at(event_type).c_str()); - if (op == "ha_scope_event") + auto key = getHaSetObjectKey(ha_set_id); + if (key.empty()) { - std::time_t now_time = getNowTime(); + SWSS_LOG_ERROR("HA Set object not found for ID: %" PRIx64, ha_set_id); + continue; + } + std::vector fvs = { + {"last_updated_time", to_string(now_time)}, + {"dp_channel_is_alive", sai_ha_set_event_type_name.at(event_type)} + }; + m_dpuStateDbHaSetTable->set(key, fvs); + } + sai_deserialize_free_ha_set_event_ntf(count, ha_set_event); +} - uint32_t count; - sai_ha_scope_event_data_t *ha_scope_event = nullptr; +void DashHaOrch::handleHaScopeEventNotification(const std::string &data) +{ + SWSS_LOG_ENTER(); - sai_deserialize_ha_scope_event_ntf(data, count, &ha_scope_event); + std::time_t now_time = getNowTime(); - for (uint32_t i = 0; i < count; i++) - { - sai_ha_scope_event_t event_type = ha_scope_event[i].event_type; - sai_object_id_t ha_scope_id = ha_scope_event[i].ha_scope_id; + uint32_t count; + sai_ha_scope_event_data_t *ha_scope_event = nullptr; - SWSS_LOG_INFO("Get HA Scope event notification id:%" PRIx64 " event: %s", ha_scope_id, sai_ha_scope_event_type_name.at(event_type).c_str()); + sai_deserialize_ha_scope_event_ntf(data, count, &ha_scope_event); - auto key = getHaScopeObjectKey(ha_scope_id); - if (key.empty()) - { - SWSS_LOG_ERROR("HA Scope object not found for ID: %" PRIx64, ha_scope_id); - continue; - } + for (uint32_t i = 0; i < count; i++) + { + sai_ha_scope_event_t event_type = ha_scope_event[i].event_type; + sai_object_id_t ha_scope_id = ha_scope_event[i].ha_scope_id; - std::vector fvs = { - {"last_updated_time", to_string(now_time)}, - {"ha_term", to_string(ha_scope_event[i].flow_version)} - }; + SWSS_LOG_INFO("Get HA Scope event notification id:%" PRIx64 " event: %s", ha_scope_id, sai_ha_scope_event_type_name.at(event_type).c_str()); - auto ha_role = to_pb(ha_scope_event[i].ha_role); - std::time_t role_start_time = now_time; + auto key = getHaScopeObjectKey(ha_scope_id); + if (key.empty()) + { + SWSS_LOG_ERROR("HA Scope object not found for ID: %" PRIx64, ha_scope_id); + continue; + } + + std::vector fvs = { + {"last_updated_time", to_string(now_time)}, + {"ha_term", to_string(ha_scope_event[i].flow_version)} + }; + + auto ha_role = to_pb(ha_scope_event[i].ha_role); + std::time_t role_start_time = now_time; - if (m_ha_scope_entries[key].metadata.ha_role() != ha_role) + if (m_ha_scope_entries[key].metadata.ha_role() != ha_role) + { + m_ha_scope_entries[key].metadata.set_ha_role(ha_role); + m_ha_scope_entries[key].last_role_start_time = now_time; + SWSS_LOG_NOTICE("HA Scope role changed for %s to %s", key.c_str(), dash::types::HaRole_Name(ha_role).c_str()); + } else + { + role_start_time = m_ha_scope_entries[key].last_role_start_time; + } + + fvs.push_back({"ha_role", sai_ha_role_name.at(ha_scope_event[i].ha_role)}); + fvs.push_back({"ha_role_start_time", to_string(role_start_time)}); + + switch (event_type) + { + case SAI_HA_SCOPE_EVENT_FLOW_RECONCILE_NEEDED: + fvs.push_back({"flow_reconcile_pending", "true"}); + break; + case SAI_HA_SCOPE_EVENT_SPLIT_BRAIN_DETECTED: + fvs.push_back({"brainsplit_recover_pending", "true"}); + break; + case SAI_HA_SCOPE_EVENT_STATE_CHANGED: + if (in(ha_scope_event[i].ha_state, {SAI_DASH_HA_STATE_PENDING_STANDALONE_ACTIVATION, + SAI_DASH_HA_STATE_PENDING_ACTIVE_ACTIVATION, + SAI_DASH_HA_STATE_PENDING_STANDBY_ACTIVATION})) { - m_ha_scope_entries[key].metadata.set_ha_role(ha_role); - m_ha_scope_entries[key].last_role_start_time = now_time; - SWSS_LOG_NOTICE("HA Scope role changed for %s to %s", key.c_str(), dash::types::HaRole_Name(ha_role).c_str()); - } else + fvs.push_back({"activate_role_pending", "true"}); + SWSS_LOG_NOTICE("DPU is pending on role activation for %s", key.c_str()); + } + else if (in(ha_scope_event[i].ha_state, {SAI_DASH_HA_STATE_ACTIVE, + SAI_DASH_HA_STATE_STANDBY})) { - role_start_time = m_ha_scope_entries[key].last_role_start_time; + fvs.push_back({"brainsplit_recover_pending", "false"}); } - fvs.push_back({"ha_role", sai_ha_role_name.at(ha_scope_event[i].ha_role)}); - fvs.push_back({"ha_role_start_time", to_string(role_start_time)}); + fvs.push_back({"ha_state", sai_ha_state_name.at(ha_scope_event[i].ha_state)}); + fvs.push_back({"ha_state_start_time", to_string(now_time)}); - switch (event_type) + m_ha_scope_entries[key].ha_state = ha_scope_event[i].ha_state; + m_ha_scope_entries[key].last_state_start_time = now_time; + + if (has_dpu_scope() && in(ha_scope_event[i].ha_state, {SAI_DASH_HA_STATE_ACTIVE, + SAI_DASH_HA_STATE_STANDBY, + SAI_DASH_HA_STATE_STANDALONE})) { - case SAI_HA_SCOPE_EVENT_FLOW_RECONCILE_NEEDED: - fvs.push_back({"flow_reconcile_pending", "true"}); - break; - case SAI_HA_SCOPE_EVENT_SPLIT_BRAIN_DETECTED: - fvs.push_back({"brainsplit_recover_pending", "true"}); - break; - case SAI_HA_SCOPE_EVENT_STATE_CHANGED: - if (in(ha_scope_event[i].ha_state, {SAI_DASH_HA_STATE_PENDING_STANDALONE_ACTIVATION, - SAI_DASH_HA_STATE_PENDING_ACTIVE_ACTIVATION, - SAI_DASH_HA_STATE_PENDING_STANDBY_ACTIVATION})) - { - fvs.push_back({"activate_role_pending", "true"}); - SWSS_LOG_NOTICE("DPU is pending on role activation for %s", key.c_str()); - } - else if (in(ha_scope_event[i].ha_state, {SAI_DASH_HA_STATE_ACTIVE, - SAI_DASH_HA_STATE_STANDBY})) - { - fvs.push_back({"brainsplit_recover_pending", "false"}); - } - - fvs.push_back({"ha_state", sai_ha_state_name.at(ha_scope_event[i].ha_state)}); - fvs.push_back({"ha_state_start_time", to_string(now_time)}); - - m_ha_scope_entries[key].ha_state = ha_scope_event[i].ha_state; - m_ha_scope_entries[key].last_state_start_time = now_time; - - if (has_dpu_scope() && in(ha_scope_event[i].ha_state, {SAI_DASH_HA_STATE_ACTIVE, - SAI_DASH_HA_STATE_STANDBY, - SAI_DASH_HA_STATE_STANDALONE})) - { - processCachedBfdSessions(); - } - break; - default: - SWSS_LOG_ERROR("Unknown HA Scope event type %d for %s", event_type, key.c_str()); + processCachedBfdSessions(); } + break; + default: + SWSS_LOG_ERROR("Unknown HA Scope event type %d for %s", event_type, key.c_str()); + } - m_dpuStateDbHaScopeTable->set(key, fvs); + m_dpuStateDbHaScopeTable->set(key, fvs); - } - sai_deserialize_free_ha_scope_event_ntf(count, ha_scope_event); - } } + sai_deserialize_free_ha_scope_event_ntf(count, ha_scope_event); } bool DashHaOrch::convertKfvToHaSetPb(const std::vector &kfv, dash::ha_set::HaSet &entry) diff --git a/orchagent/dash/dashhaorch.h b/orchagent/dash/dashhaorch.h index 3a3315a5160..c7c342bc1d2 100644 --- a/orchagent/dash/dashhaorch.h +++ b/orchagent/dash/dashhaorch.h @@ -66,6 +66,9 @@ class DashHaOrch : public ZmqOrch void doTask(ConsumerBase &consumer); void doTask(swss::NotificationConsumer &consumer); + void handleNotification(swss::KeyOpFieldsValuesTuple &entry); + void handleHaSetEventNotification(const std::string &data); + void handleHaScopeEventNotification(const std::string &data); void doTaskEniTable(ConsumerBase &consumer); void doTaskHaSetTable(ConsumerBase &consumer); void doTaskHaScopeTable(ConsumerBase &consumer); diff --git a/orchagent/fdborch.cpp b/orchagent/fdborch.cpp index 389519917db..9fe907f0937 100644 --- a/orchagent/fdborch.cpp +++ b/orchagent/fdborch.cpp @@ -19,6 +19,8 @@ #include "directory.h" #include "timer.h" #include "neighorch.h" +#include "notifications.h" +#include "sainotificationorch.h" #define VLAN_PREFIX "Vlan" @@ -89,6 +91,17 @@ FdbOrch::FdbOrch(DBConnector* applDbConnector, vector app auto fdbNotifier = new Notifier(m_fdbNotificationConsumer, this, "FDB_NOTIFICATIONS"); Orch::addExecutor(fdbNotifier); + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }, + [this]() { return m_portsOrch->allPortsReady(); }); + } + /* MAC Move Guard: detects MAC flapping between ports and applies a remediation (admin-disable port, or pre-ingress ACL learn-suppress). Owned by FdbOrch via composition; its config-table Consumer and @@ -1400,32 +1413,48 @@ void FdbOrch::doTask(NotificationConsumer& consumer) } else if (&consumer == m_fdbNotificationConsumer && op == "fdb_event") { - uint32_t count; - sai_fdb_event_notification_data_t *fdbevent = nullptr; - sai_deserialize_fdb_event_ntf(data, count, &fdbevent); + KeyOpFieldsValuesTuple entry = std::make_tuple(data, op, values); + handleNotification(entry); + } +} - for (uint32_t i = 0; i < count; ++i) - { - sai_object_id_t oid = SAI_NULL_OBJECT_ID; - sai_fdb_entry_type_t sai_fdb_type = SAI_FDB_ENTRY_TYPE_DYNAMIC; +void FdbOrch::handleNotification(KeyOpFieldsValuesTuple &entry) +{ + if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT) + { + handleFdbEventNotification(kfvKey(entry)); + } +} + +void FdbOrch::handleFdbEventNotification(const std::string& data) +{ + SWSS_LOG_ENTER(); + + uint32_t count; + sai_fdb_event_notification_data_t *fdbevent = nullptr; + sai_deserialize_fdb_event_ntf(data, count, &fdbevent); + + for (uint32_t i = 0; i < count; ++i) + { + sai_object_id_t oid = SAI_NULL_OBJECT_ID; + sai_fdb_entry_type_t sai_fdb_type = SAI_FDB_ENTRY_TYPE_DYNAMIC; - for (uint32_t j = 0; j < fdbevent[i].attr_count; ++j) + for (uint32_t j = 0; j < fdbevent[i].attr_count; ++j) + { + if (fdbevent[i].attr[j].id == SAI_FDB_ENTRY_ATTR_BRIDGE_PORT_ID) { - if (fdbevent[i].attr[j].id == SAI_FDB_ENTRY_ATTR_BRIDGE_PORT_ID) - { - oid = fdbevent[i].attr[j].value.oid; - } - else if (fdbevent[i].attr[j].id == SAI_FDB_ENTRY_ATTR_TYPE) - { - sai_fdb_type = (sai_fdb_entry_type_t)fdbevent[i].attr[j].value.s32; - } + oid = fdbevent[i].attr[j].value.oid; + } + else if (fdbevent[i].attr[j].id == SAI_FDB_ENTRY_ATTR_TYPE) + { + sai_fdb_type = (sai_fdb_entry_type_t)fdbevent[i].attr[j].value.s32; } - - this->update(fdbevent[i].event_type, &fdbevent[i].fdb_entry, oid, sai_fdb_type); } - sai_deserialize_free_fdb_event_ntf(count, fdbevent); + this->update(fdbevent[i].event_type, &fdbevent[i].fdb_entry, oid, sai_fdb_type); } + + sai_deserialize_free_fdb_event_ntf(count, fdbevent); } /* diff --git a/orchagent/fdborch.h b/orchagent/fdborch.h index c79634c3de5..8cdd762e3ff 100644 --- a/orchagent/fdborch.h +++ b/orchagent/fdborch.h @@ -171,6 +171,8 @@ class FdbOrch: public Orch, public Subject, public Observer void doTask(Consumer& consumer); void doTask(NotificationConsumer& consumer); void doTask(swss::SelectableTimer& timer) override; + void handleNotification(swss::KeyOpFieldsValuesTuple &entry); + void handleFdbEventNotification(const std::string& data); void updateVlanMember(const VlanMemberUpdate&); void updatePortOperState(const PortOperStateUpdate&); diff --git a/orchagent/high_frequency_telemetry/hftelorch.cpp b/orchagent/high_frequency_telemetry/hftelorch.cpp index 8c4807f6663..20a912ca8e2 100644 --- a/orchagent/high_frequency_telemetry/hftelorch.cpp +++ b/orchagent/high_frequency_telemetry/hftelorch.cpp @@ -2,6 +2,7 @@ #include "hftelutils.h" #include "notifications.h" +#include "sainotificationorch.h" #include #include @@ -90,6 +91,16 @@ HFTelOrch::HFTelOrch( } Orch::addExecutor(notifier); + + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_TAM_TEL_TYPE_CONFIG_CHANGE, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + } } HFTelOrch::~HFTelOrch() @@ -504,11 +515,25 @@ void HFTelOrch::doTask(swss::NotificationConsumer &consumer) consumer.pop(op, data, values); - if (op != SAI_SWITCH_NOTIFICATION_NAME_TAM_TEL_TYPE_CONFIG_CHANGE) + KeyOpFieldsValuesTuple entry = std::make_tuple(data, op, values); + handleNotification(entry); +} + +void HFTelOrch::handleNotification(KeyOpFieldsValuesTuple &entry) +{ + if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_TAM_TEL_TYPE_CONFIG_CHANGE) { - SWSS_LOG_DEBUG("Unknown operation type %s for HFTel Orch", op.c_str()); - return; + handleTamTelTypeConfigChangeNotification(kfvKey(entry)); + } + else + { + SWSS_LOG_DEBUG("Unknown operation type %s for HFTel Orch", kfvOp(entry).c_str()); } +} + +void HFTelOrch::handleTamTelTypeConfigChangeNotification(const std::string &data) +{ + SWSS_LOG_ENTER(); sai_object_id_t tam_tel_type_obj = SAI_NULL_OBJECT_ID; diff --git a/orchagent/high_frequency_telemetry/hftelorch.h b/orchagent/high_frequency_telemetry/hftelorch.h index 845a18fa156..5381f5afc13 100644 --- a/orchagent/high_frequency_telemetry/hftelorch.h +++ b/orchagent/high_frequency_telemetry/hftelorch.h @@ -49,7 +49,9 @@ class HFTelOrch : public Orch bool isProfileInUse(const std::shared_ptr &profile) const; void doTask(swss::NotificationConsumer &consumer); + void handleNotification(swss::KeyOpFieldsValuesTuple &entry); void doTask(Consumer &consumer); + void handleTamTelTypeConfigChangeNotification(const std::string &data); // SAI objects sai_object_id_t m_sai_hostif_obj; diff --git a/orchagent/icmporch.cpp b/orchagent/icmporch.cpp index 7fd252d6d86..777f37352a5 100644 --- a/orchagent/icmporch.cpp +++ b/orchagent/icmporch.cpp @@ -12,6 +12,7 @@ #include "sai_serialize.h" #include "directory.h" #include "notifications.h" +#include "sainotificationorch.h" #include "icmporch.h" #include "switchorch.h" #include @@ -21,6 +22,7 @@ using namespace std; using namespace swss; extern SwitchOrch *gSwitchOrch; +extern sai_redis_communication_mode_t gRedisCommunicationMode; const uint32_t IcmpOrch::m_max_sessions = 1024; @@ -72,6 +74,18 @@ IcmpOrch::IcmpOrch(DBConnector *db, string tableName, TableConnector stateDbIcmp auto icmpStateNotifier = new Notifier(m_icmpStateNotificationConsumer, this, "ICMP_STATE_NOTIFICATIONS"); Orch::addExecutor(icmpStateNotifier); + Orch::addExecutor(icmpStateNotifier); + + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_ICMP_ECHO_SESSION_STATE_CHANGE, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + } + initializeCounters(); } @@ -200,47 +214,58 @@ void IcmpOrch::doTask(NotificationConsumer &consumer) return; } - if (op == "icmp_echo_session_state_change") + KeyOpFieldsValuesTuple entry = std::make_tuple(data, op, values); + handleNotification(entry); +} + +void IcmpOrch::handleNotification(KeyOpFieldsValuesTuple &entry) +{ + if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_ICMP_ECHO_SESSION_STATE_CHANGE) { - uint32_t count = 0; - sai_icmp_echo_session_state_notification_t *icmpSessionState = nullptr; + handleIcmpEchoSessionStateChangeNotification(kfvKey(entry)); + } +} - sai_deserialize_icmp_echo_session_state_ntf(data, count, &icmpSessionState); +void IcmpOrch::handleIcmpEchoSessionStateChangeNotification(const std::string &data) +{ + uint32_t count = 0; + sai_icmp_echo_session_state_notification_t *icmpSessionState = nullptr; - for (uint32_t i = 0; i < count; i++) - { - sai_object_id_t id = icmpSessionState[i].icmp_echo_session_id; - sai_icmp_echo_session_state_t state = icmpSessionState[i].session_state; + sai_deserialize_icmp_echo_session_state_ntf(data, count, &icmpSessionState); - SWSS_LOG_INFO("Got ICMP session state change notification id:%" PRIx64 " state: %s", id, m_session_state_lkup.at(state).c_str()); + for (uint32_t i = 0; i < count; i++) + { + sai_object_id_t id = icmpSessionState[i].icmp_echo_session_id; + sai_icmp_echo_session_state_t state = icmpSessionState[i].session_state; - if (m_icmp_session_lookup.find(id) == m_icmp_session_lookup.end()) - { - SWSS_LOG_NOTICE("ICMP session missing for state change notification id:%" PRIx64 " state: %s", id, m_session_state_lkup.at(state).c_str()); - continue; - } + SWSS_LOG_INFO("Got ICMP session state change notification id:%" PRIx64 " state: %s", id, m_session_state_lkup.at(state).c_str()); - // handle state update - if (state != m_icmp_session_lookup[id].state || m_icmp_session_lookup[id].init_state) - { - auto key = m_icmp_session_lookup[id].db_key; - vector fvVector; - m_stateIcmpSessionTable.get(key, fvVector); + if (m_icmp_session_lookup.find(id) == m_icmp_session_lookup.end()) + { + SWSS_LOG_NOTICE("ICMP session missing for state change notification id:%" PRIx64 " state: %s", id, m_session_state_lkup.at(state).c_str()); + continue; + } - fvVector.push_back({IcmpSaiSessionHandler::m_state_fname, m_session_state_lkup.at(state)}); + // handle state update + if (state != m_icmp_session_lookup[id].state || m_icmp_session_lookup[id].init_state) + { + auto key = m_icmp_session_lookup[id].db_key; + vector fvVector; + m_stateIcmpSessionTable.get(key, fvVector); - m_stateIcmpSessionTable.set(key, fvVector); + fvVector.push_back({IcmpSaiSessionHandler::m_state_fname, m_session_state_lkup.at(state)}); - SWSS_LOG_NOTICE("ICMP session state for %s changed from %s to %s", key.c_str(), - m_session_state_lkup.at(m_icmp_session_lookup[id].state).c_str(), m_session_state_lkup.at(state).c_str()); + m_stateIcmpSessionTable.set(key, fvVector); - m_icmp_session_lookup[id].state = state; - m_icmp_session_lookup[id].init_state = false; - } - } + SWSS_LOG_NOTICE("ICMP session state for %s changed from %s to %s", key.c_str(), + m_session_state_lkup.at(m_icmp_session_lookup[id].state).c_str(), m_session_state_lkup.at(state).c_str()); - sai_deserialize_free_icmp_echo_session_state_ntf(count, icmpSessionState); + m_icmp_session_lookup[id].state = state; + m_icmp_session_lookup[id].init_state = false; + } } + + sai_deserialize_free_icmp_echo_session_state_ntf(count, icmpSessionState); } bool IcmpOrch::create_icmp_session(const string& key, const vector& data) @@ -734,14 +759,12 @@ SaiOffloadHandlerStatus IcmpSaiSessionHandler::do_update() void IcmpSaiSessionHandler::on_state_change(uint32_t count, sai_icmp_echo_session_state_notification_t *data) { - extern sai_redis_communication_mode_t gRedisCommunicationMode; if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - static thread_local swss::DBConnector db("ASIC_DB", 0); - static thread_local swss::NotificationProducer icmpNotifier(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_icmp_echo_session_state_ntf(count, data); std::vector values; - icmpNotifier.send("icmp_echo_session_state_change", sdata, values); + + enqueueSaiNotification("icmp_echo_session_state_change", std::move(sdata), std::move(values)); } } diff --git a/orchagent/icmporch.h b/orchagent/icmporch.h index c1c4af94727..afbd2f15014 100644 --- a/orchagent/icmporch.h +++ b/orchagent/icmporch.h @@ -169,6 +169,8 @@ class IcmpOrch: public Orch, public Subject * true for all other cases where session entry is consumed */ bool update_icmp_session(const string& key, const vector& data); + void handleNotification(swss::KeyOpFieldsValuesTuple &entry); + void handleIcmpEchoSessionStateChangeNotification(const std::string &data); /** *@method initializeCounters diff --git a/orchagent/macsecorch.cpp b/orchagent/macsecorch.cpp index 51febf62ea9..670317ae359 100644 --- a/orchagent/macsecorch.cpp +++ b/orchagent/macsecorch.cpp @@ -1,6 +1,8 @@ #include "macsecorch.h" #include "macsecpost.h" #include "notifier.h" +#include "sainotificationorch.h" +#include "saiextensions.h" #include #include @@ -690,6 +692,22 @@ MACsecOrch::MACsecOrch( m_postCompletionNotificationConsumer = new swss::NotificationConsumer(m_notificationsDb.get(), "NOTIFICATIONS"); auto postCompletionNotificatier = new Notifier(m_postCompletionNotificationConsumer, this, "POST_COMPLETION__NOTIFICATIONS"); Orch::addExecutor(postCompletionNotificatier); + + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_SWITCH_MACSEC_POST_STATUS, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_MACSEC_POST_STATUS, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + } } if (post_state == "switch-level-post-in-progress") @@ -752,24 +770,24 @@ void MACsecOrch::doTask(NotificationConsumer &consumer) consumer.pops(entries); for (auto& entry : entries) { - handleNotification(consumer, entry); + handleNotification(entry); } } -void MACsecOrch::handleNotification(NotificationConsumer &consumer, KeyOpFieldsValuesTuple& entry) +void MACsecOrch::handleNotification(KeyOpFieldsValuesTuple& entry) { SWSS_LOG_ENTER(); - if (&consumer != m_postCompletionNotificationConsumer) - { - return; - } + handleMacsecPostNotification(kfvOp(entry), kfvKey(entry)); +} + +void MACsecOrch::handleMacsecPostNotification(const std::string &op, const std::string &data) +{ + SWSS_LOG_ENTER(); - auto op = kfvOp(entry); - auto data = kfvKey(entry); SWSS_LOG_NOTICE("Received SAI notification: op %s, data %s", op.c_str(), data.c_str()); - if (op == "switch_macsec_post_status") + if (op == SAI_SWITCH_NOTIFICATION_NAME_SWITCH_MACSEC_POST_STATUS) { sai_object_id_t switch_id; sai_switch_macsec_post_status_t switch_macsec_post_status; @@ -798,7 +816,7 @@ void MACsecOrch::handleNotification(NotificationConsumer &consumer, KeyOpFieldsV } } - if (op == "macsec_post_status") + if (op == SAI_SWITCH_NOTIFICATION_NAME_MACSEC_POST_STATUS) { sai_object_id_t macsec_id; sai_macsec_post_status_t macsec_post_status; diff --git a/orchagent/macsecorch.h b/orchagent/macsecorch.h index 6673f701018..603307ac4ae 100644 --- a/orchagent/macsecorch.h +++ b/orchagent/macsecorch.h @@ -41,7 +41,8 @@ class MACsecOrch : public Orch private: void doTask(Consumer &consumer); void doTask(NotificationConsumer &consumer); - void handleNotification(NotificationConsumer &consumer, KeyOpFieldsValuesTuple& entry); + void handleNotification(swss::KeyOpFieldsValuesTuple& entry); + void handleMacsecPostNotification(const std::string &op, const std::string &data); public: using TaskArgs = std::vector; diff --git a/orchagent/notifications.cpp b/orchagent/notifications.cpp index 525bc680e56..f5ebdd1ee2d 100644 --- a/orchagent/notifications.cpp +++ b/orchagent/notifications.cpp @@ -3,44 +3,304 @@ extern "C" { } #include "logger.h" +#include "orch.h" +#include "notificationconsumer.h" #include "notifications.h" +#include "sai_serialize.h" #include "switchorch.h" +#include +#include +#include +#include + extern SwitchOrch *gSwitchOrch; extern sai_redis_communication_mode_t gRedisCommunicationMode; +volatile sig_atomic_t gOrchShutdownRequested = 0; + +static SaiNotificationQueue *gSaiNotificationQueue = nullptr; +static SaiNotificationDispatcher *gSaiNotificationDispatcher = nullptr; #ifdef ASAN_ENABLED #include #endif +class SaiNotificationQueueSelectable : public swss::Selectable +{ +public: + explicit SaiNotificationQueueSelectable(SaiNotificationQueue *queue) + : m_queue(queue) + { + } + + int getFd() override + { + return m_queue->getFd(); + } + + uint64_t readData() override + { + return m_queue->readData(); + } + + bool hasData() override + { + return m_queue->hasData(); + } + + bool hasCachedData() override + { + return m_queue->hasCachedData(); + } + +private: + SaiNotificationQueue *m_queue; +}; + +class SaiNotificationQueueExecutor : public Executor +{ +public: + SaiNotificationQueueExecutor(SaiNotificationQueue *queue, + Orch *orch, + SaiNotificationDispatcher *dispatcher, + const std::string &name); + + void execute() override; + void drain() override; + +private: + SaiNotificationQueue *m_queue; + SaiNotificationDispatcher *m_dispatcher; +}; + +SaiNotificationQueue::SaiNotificationQueue(int pri, size_t popBatchSize) + : swss::Selectable(pri) + , m_selectableEvent(pri) + , m_popBatchSize(popBatchSize) +{ +} + +void SaiNotificationQueue::enqueue(const std::string &op, std::string data, std::vector values) +{ + { + std::lock_guard lock(m_mutex); + m_queue.emplace(std::move(data), op, std::move(values)); + if (m_queue.size() > m_highWatermark) + { + m_highWatermark = m_queue.size(); + } + } + + m_selectableEvent.notify(); +} + +size_t SaiNotificationQueue::size() const +{ + std::lock_guard lock(m_mutex); + return m_queue.size(); +} + +size_t SaiNotificationQueue::highWatermark() const +{ + std::lock_guard lock(m_mutex); + return m_highWatermark; +} + +bool SaiNotificationQueue::peekFrontOp(std::string &op) const +{ + std::lock_guard lock(m_mutex); + if (m_queue.empty()) + { + return false; + } + + op = kfvOp(m_queue.front()); + return true; +} + +void SaiNotificationQueue::pops(std::deque &entries) +{ + entries.clear(); + + std::lock_guard lock(m_mutex); + const auto count = std::min(m_queue.size(), m_popBatchSize); + for (size_t i = 0; i < count; ++i) + { + entries.push_back(std::move(m_queue.front())); + m_queue.pop(); + } + + if (!m_queue.empty()) + { + m_selectableEvent.notify(); + } +} + +int SaiNotificationQueue::getFd() +{ + return m_selectableEvent.getFd(); +} + +uint64_t SaiNotificationQueue::readData() +{ + return m_selectableEvent.readData(); +} + +bool SaiNotificationQueue::hasData() +{ + std::lock_guard lock(m_mutex); + return !m_queue.empty(); +} + +bool SaiNotificationQueue::hasCachedData() +{ + return hasData(); +} + +void SaiNotificationDispatcher::registerHandler(const std::string &op, Handler handler, + ReadinessPredicate ready) +{ + std::lock_guard lock(m_mutex); + m_handlers[op] = std::move(handler); + m_readiness[op] = std::move(ready); +} + +bool SaiNotificationDispatcher::isReady(const std::string &op) const +{ + std::lock_guard lock(m_mutex); + auto readyIt = m_readiness.find(op); + if (readyIt == m_readiness.end() || !readyIt->second) + { + return true; + } + + return readyIt->second(); +} + +void SaiNotificationDispatcher::dispatch(swss::KeyOpFieldsValuesTuple &entry) +{ + Handler handler; + auto op = kfvOp(entry); + + { + std::lock_guard lock(m_mutex); + auto handlerIt = m_handlers.find(op); + if (handlerIt != m_handlers.end()) + { + handler = handlerIt->second; + } + } + + if (handler) + { + handler(entry); + } + else + { + SWSS_LOG_WARN("No handler registered for SAI notification op %s", op.c_str()); + } +} + +SaiNotificationQueueExecutor::SaiNotificationQueueExecutor(SaiNotificationQueue *queue, + Orch *orch, + SaiNotificationDispatcher *dispatcher, + const std::string &name) + : Executor(new SaiNotificationQueueSelectable(queue), orch, name) + , m_queue(queue) + , m_dispatcher(dispatcher) +{ +} + +void SaiNotificationQueueExecutor::execute() +{ + std::string frontOp; + if (!m_queue->peekFrontOp(frontOp) || !m_dispatcher->isReady(frontOp)) + { + return; + } + + std::deque entries; + m_queue->pops(entries); + + for (auto &entry : entries) + { + m_dispatcher->dispatch(entry); + } +} + +void SaiNotificationQueueExecutor::drain() +{ + execute(); +} + +Executor *createSaiNotificationQueueExecutor(SaiNotificationQueue *queue, + Orch *orch, + SaiNotificationDispatcher *dispatcher, + const std::string &name) +{ + return new SaiNotificationQueueExecutor(queue, orch, dispatcher, name); +} + +SaiNotificationQueue *getSaiNotificationQueue() +{ + static std::mutex queueMutex; + + std::lock_guard lock(queueMutex); + if (gSaiNotificationQueue == nullptr) + { + gSaiNotificationQueue = new SaiNotificationQueue(100, swss::DEFAULT_NC_POP_BATCH_SIZE); + } + + return gSaiNotificationQueue; +} + +SaiNotificationDispatcher *getSaiNotificationDispatcher() +{ + static std::mutex dispatcherMutex; + + std::lock_guard lock(dispatcherMutex); + if (gSaiNotificationDispatcher == nullptr) + { + gSaiNotificationDispatcher = new SaiNotificationDispatcher(); + } + + return gSaiNotificationDispatcher; +} + +void enqueueSaiNotification(const std::string &op, std::string data, std::vector values) +{ + if (gOrchShutdownRequested != 0) + { + return; + } + + getSaiNotificationQueue()->enqueue(op, std::move(data), std::move(values)); +} + void on_fdb_event(uint32_t count, sai_fdb_event_notification_data_t *data) { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - static thread_local swss::DBConnector db("ASIC_DB", 0); - static thread_local swss::NotificationProducer fdbNotifier(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_fdb_event_ntf(count, data); std::vector values; - fdbNotifier.send("fdb_event", sdata, values); + + enqueueSaiNotification("fdb_event", std::move(sdata), std::move(values)); } } /* * Don't perform DB operations within this event handler, because it runs by * libsairedis in a separate thread which causes concurrency issues. - * For platforms which use zmq between orchagent and syncd, it is an acceptable - * workaround to forward the notifications from the callback handler to the - * redis notifications channel processed by portsorch. + * In ZMQ mode, enqueue the notification so orchagent's main loop can process it. */ void on_port_state_change(uint32_t count, sai_port_oper_status_notification_t *data) { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - static thread_local swss::DBConnector db("ASIC_DB", 0); - static thread_local swss::NotificationProducer portStateNotifier(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_port_oper_status_ntf(count, data); std::vector values; - portStateNotifier.send("port_state_change", sdata, values); + + enqueueSaiNotification("port_state_change", std::move(sdata), std::move(values)); } } @@ -48,11 +308,10 @@ void on_bfd_session_state_change(uint32_t count, sai_bfd_session_state_notificat { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - static thread_local swss::DBConnector db("ASIC_DB", 0); - static thread_local swss::NotificationProducer bfdNotifier(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_bfd_session_state_ntf(count, data); std::vector values; - bfdNotifier.send("bfd_session_state_change", sdata, values); + + enqueueSaiNotification("bfd_session_state_change", std::move(sdata), std::move(values)); } } @@ -60,11 +319,10 @@ void on_twamp_session_event(uint32_t count, sai_twamp_session_event_notification { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - static thread_local swss::DBConnector db("ASIC_DB", 0); - static thread_local swss::NotificationProducer twampNotifier(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_twamp_session_event_ntf(count, data); std::vector values; - twampNotifier.send("twamp_session_event", sdata, values); + + enqueueSaiNotification("twamp_session_event", std::move(sdata), std::move(values)); } } @@ -72,13 +330,10 @@ void on_ha_set_event(uint32_t count, sai_ha_set_event_data_t *data) { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - swss::DBConnector db("ASIC_DB", 0); - swss::NotificationProducer ha_set_event(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_ha_set_event_ntf(count, data); std::vector values; - // Forward ha_set_event notification to be handled in dashhaorch doTask() - ha_set_event.send(SAI_SWITCH_NOTIFICATION_NAME_HA_SET_EVENT, sdata, values); + enqueueSaiNotification(SAI_SWITCH_NOTIFICATION_NAME_HA_SET_EVENT, std::move(sdata), std::move(values)); } } @@ -86,13 +341,10 @@ void on_ha_scope_event(uint32_t count, sai_ha_scope_event_data_t *data) { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - swss::DBConnector db("ASIC_DB", 0); - swss::NotificationProducer ha_scope_event(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_ha_scope_event_ntf(count, data); std::vector values; - // Forward ha_scope_event notification to be handled in dashhaorch doTask() - ha_scope_event.send(SAI_SWITCH_NOTIFICATION_NAME_HA_SCOPE_EVENT, sdata, values); + enqueueSaiNotification(SAI_SWITCH_NOTIFICATION_NAME_HA_SCOPE_EVENT, std::move(sdata), std::move(values)); } } @@ -100,13 +352,10 @@ void on_flow_bulk_get_session_event(sai_object_id_t flow_bulk_session_id, uint32 { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - swss::DBConnector db("ASIC_DB", 0); - swss::NotificationProducer flow_bulk_get_session_event(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_flow_bulk_get_session_event_ntf(flow_bulk_session_id, count, data); std::vector values; - // Forward flow_bulk_get_session_event notification to be handled in orchagent doTask() - flow_bulk_get_session_event.send(SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT, sdata, values); + enqueueSaiNotification(SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT, std::move(sdata), std::move(values)); } } @@ -137,15 +386,14 @@ void on_switch_shutdown_request(sai_object_id_t switch_id) quick_exit(EXIT_FAILURE); } -void on_port_host_tx_ready(sai_object_id_t switch_id, sai_object_id_t port_id, sai_port_host_tx_ready_status_t m_portHostTxReadyStatus) +void on_port_host_tx_ready(sai_object_id_t switch_id, sai_object_id_t port_id, sai_port_host_tx_ready_status_t hostTxReadyStatus) { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - static thread_local swss::DBConnector db("ASIC_DB", 0); - static thread_local swss::NotificationProducer portHostTxReadyNotifier(&db, "NOTIFICATIONS"); - std::string sdata = sai_serialize_port_host_tx_ready_ntf(switch_id, port_id, m_portHostTxReadyStatus); + std::string sdata = sai_serialize_port_host_tx_ready_ntf(switch_id, port_id, hostTxReadyStatus); std::vector values; - portHostTxReadyNotifier.send("port_host_tx_ready", sdata, values); + + enqueueSaiNotification("port_host_tx_ready", std::move(sdata), std::move(values)); } } @@ -168,11 +416,10 @@ void on_tam_tel_type_config_change(sai_object_id_t tam_tel_id) { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - static thread_local swss::DBConnector db("ASIC_DB", 0); - static thread_local swss::NotificationProducer tamNotifier(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_object_id(tam_tel_id); std::vector values; - tamNotifier.send(SAI_SWITCH_NOTIFICATION_NAME_TAM_TEL_TYPE_CONFIG_CHANGE, sdata, values); + + enqueueSaiNotification(SAI_SWITCH_NOTIFICATION_NAME_TAM_TEL_TYPE_CONFIG_CHANGE, std::move(sdata), std::move(values)); } } @@ -181,13 +428,10 @@ void on_switch_macsec_post_status_notify(sai_object_id_t switch_id, { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - swss::DBConnector db("ASIC_DB", 0); - swss::NotificationProducer macsec_post_status_notify(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_switch_macsec_post_status_ntf(switch_id, switch_macsec_post_status); std::vector values; - // Forward switch_macsec_post_status notification to be handled in macsecorch doTask() - macsec_post_status_notify.send("switch_macsec_post_status", sdata, values); + enqueueSaiNotification(SAI_SWITCH_NOTIFICATION_NAME_SWITCH_MACSEC_POST_STATUS, std::move(sdata), std::move(values)); } } @@ -196,12 +440,9 @@ void on_macsec_post_status_notify(sai_object_id_t macsec_id, { if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) { - swss::DBConnector db("ASIC_DB", 0); - swss::NotificationProducer macsec_post_status_notify(&db, "NOTIFICATIONS"); std::string sdata = sai_serialize_macsec_post_status_ntf(macsec_id, macsec_post_status); std::vector values; - // Forward macsec_post_status notification to be handled in macsecorch doTask() - macsec_post_status_notify.send("macsec_post_status", sdata, values); + enqueueSaiNotification(SAI_SWITCH_NOTIFICATION_NAME_MACSEC_POST_STATUS, std::move(sdata), std::move(values)); } } diff --git a/orchagent/notifications.h b/orchagent/notifications.h index af511212880..81991a11c81 100644 --- a/orchagent/notifications.h +++ b/orchagent/notifications.h @@ -1,10 +1,74 @@ #pragma once +#include +#include +#include +#include +#include +#include +#include +#include + extern "C" { #include "sai.h" #include "saiextensions.h" } +#include "selectableevent.h" +#include "table.h" + +class Orch; +class Executor; + +class SaiNotificationQueue : public swss::Selectable +{ +public: + SaiNotificationQueue(int pri = 100, size_t popBatchSize = 128); + + void enqueue(const std::string &op, std::string data, std::vector values); + bool peekFrontOp(std::string &op) const; + void pops(std::deque &entries); + size_t size() const; + size_t highWatermark() const; + + int getFd() override; + uint64_t readData() override; + bool hasData() override; + bool hasCachedData() override; + +private: + mutable std::mutex m_mutex; + std::queue m_queue; + swss::SelectableEvent m_selectableEvent; + size_t m_popBatchSize; + size_t m_highWatermark = 0; +}; + +class SaiNotificationDispatcher +{ +public: + using Handler = std::function; + using ReadinessPredicate = std::function; + + void registerHandler(const std::string &op, Handler handler, + ReadinessPredicate ready = nullptr); + bool isReady(const std::string &op) const; + void dispatch(swss::KeyOpFieldsValuesTuple &entry); + +private: + mutable std::mutex m_mutex; + std::unordered_map m_handlers; + std::unordered_map m_readiness; +}; + +SaiNotificationQueue *getSaiNotificationQueue(); +SaiNotificationDispatcher *getSaiNotificationDispatcher(); +void enqueueSaiNotification(const std::string &op, std::string data, std::vector values); +Executor *createSaiNotificationQueueExecutor(SaiNotificationQueue *queue, + Orch *orch, + SaiNotificationDispatcher *dispatcher, + const std::string &name); + void on_fdb_event(uint32_t count, sai_fdb_event_notification_data_t *data); void on_port_state_change(uint32_t count, sai_port_oper_status_notification_t *data); void on_bfd_session_state_change(uint32_t count, sai_bfd_session_state_notification_t *data); diff --git a/orchagent/orchdaemon.cpp b/orchagent/orchdaemon.cpp index dd048fbe115..86a5660da6e 100644 --- a/orchagent/orchdaemon.cpp +++ b/orchagent/orchdaemon.cpp @@ -17,6 +17,7 @@ #include "sairedis.h" #include "chassisorch.h" #include "notificationconsumerstatsorch.h" +#include "sainotificationorch.h" #include "stporch.h" using namespace std; @@ -31,9 +32,10 @@ using namespace swss; extern sai_switch_api_t* sai_switch_api; extern sai_object_id_t gSwitchId; +extern sai_redis_communication_mode_t gRedisCommunicationMode; extern string gMySwitchType; extern string gMySwitchSubType; -volatile sig_atomic_t gOrchShutdownRequested = 0; +extern volatile sig_atomic_t gOrchShutdownRequested; extern void syncd_apply_view(); /* @@ -243,6 +245,11 @@ bool OrchDaemon::init() { APP_MCLAG_FDB_TABLE_NAME, FdbOrch::fdborch_pri} }; + if (gRedisCommunicationMode == SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC) + { + gSaiNotificationOrch = new SaiNotificationOrch(); + } + gPortsOrch = new PortsOrch(m_applDb, m_stateDb, ports_tables, m_chassisAppDb); // Create EvpnMhOrch early so its ES/DF state is available when PortsOrch @@ -534,6 +541,10 @@ bool OrchDaemon::init() * For cases when Orch has to process tables in specific order, like PortsOrch during warm start, it has to override Orch::doTask() */ m_orchList = { gSwitchOrch, gCrmOrch, gPortsOrch, gEvpnMhOrch, gBufferOrch, gFlowCounterRouteOrch, gIntfsOrch, gNeighOrch, gNhgMapOrch, gNhgOrch, gCbfNhgOrch, gFgNhgOrch, gRouteOrch, gCoppOrch, gQosOrch, wm_orch, gPolicerOrch, gTunneldecapOrch, sflow_orch, gDebugCounterOrch, gMacsecOrch, bgp_global_state_orch, gBfdOrch, gIcmpOrch, gSrv6Orch, gMuxOrch, mux_cb_orch, gMonitorOrch, gBfdMonitorOrch, gStpOrch, gL2NhgOrch, gNotifConsumerStatsOrch}; + if (gSaiNotificationOrch) + { + m_orchList.push_back(gSaiNotificationOrch); + } bool initialize_dtel = false; if (platform == BFN_PLATFORM_SUBSTRING || platform == VS_PLATFORM_SUBSTRING) { diff --git a/orchagent/portsorch.cpp b/orchagent/portsorch.cpp index e684119d941..0e1d08941cb 100644 --- a/orchagent/portsorch.cpp +++ b/orchagent/portsorch.cpp @@ -12,6 +12,7 @@ #include "directory.h" #include "subintf.h" #include "notifications.h" +#include "sainotificationorch.h" #include "stporch.h" #include @@ -1116,6 +1117,25 @@ PortsOrch::PortsOrch(DBConnector *db, DBConnector *stateDb, vectorregisterHandler( + SAI_SWITCH_NOTIFICATION_NAME_PORT_STATE_CHANGE, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }, + [this]() { return allPortsReady(); }); + + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_PORT_HOST_TX_READY, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }, + [this]() { return allPortsReady(); }); + } + if (m_cmisModuleAsicSyncSupported) { m_portHostTxReadyNotificationConsumer = new swss::NotificationConsumer( @@ -9884,99 +9904,108 @@ void PortsOrch::doTask(NotificationConsumer &consumer) for (auto& entry : entries) { - handleNotification(consumer, entry); + handleNotification(entry); } } -void PortsOrch::handleNotification(NotificationConsumer &consumer, KeyOpFieldsValuesTuple& entry) +void PortsOrch::handleNotification(KeyOpFieldsValuesTuple& entry) { auto op = kfvOp(entry); auto data = kfvKey(entry); - auto values = kfvFieldsValues(entry); - if (&consumer == m_portStatusNotificationConsumer && op == "port_state_change") + if (op == SAI_SWITCH_NOTIFICATION_NAME_PORT_STATE_CHANGE) + { + handlePortStateChangeNotification(data); + } + else if (op == SAI_SWITCH_NOTIFICATION_NAME_PORT_HOST_TX_READY) { - uint32_t count; - sai_port_oper_status_notification_t *portoperstatus = nullptr; + handlePortHostTxReadyNotification(data); + } +} - sai_deserialize_port_oper_status_ntf(data, count, &portoperstatus); +void PortsOrch::handlePortStateChangeNotification(const std::string &data) +{ + uint32_t count; + sai_port_oper_status_notification_t *portoperstatus = nullptr; - for (uint32_t i = 0; i < count; i++) - { - Port port; - sai_object_id_t id = portoperstatus[i].port_id; - sai_port_oper_status_t status = portoperstatus[i].port_state; - sai_port_error_status_t port_oper_err = portoperstatus[i].port_error_status; + sai_deserialize_port_oper_status_ntf(data, count, &portoperstatus); - SWSS_LOG_NOTICE("Get port state change notification id:%" PRIx64 " status:%d " - "oper_error_status:0x%" PRIx32, - id, status, port_oper_err); + for (uint32_t i = 0; i < count; i++) + { + Port port; + sai_object_id_t id = portoperstatus[i].port_id; + sai_port_oper_status_t status = portoperstatus[i].port_state; + sai_port_error_status_t port_oper_err = portoperstatus[i].port_error_status; - if (!getPort(id, port)) + SWSS_LOG_NOTICE("Get port state change notification id:%" PRIx64 " status:%d " + "oper_error_status:0x%" PRIx32, + id, status, port_oper_err); + + if (!getPort(id, port)) + { + SWSS_LOG_NOTICE("Got port state change for port id 0x%" PRIx64 " which does not exist, possibly outdated event", id); + continue; + } + + updatePortOperStatus(port, status); + if (status == SAI_PORT_OPER_STATUS_UP) + { + sai_uint32_t speed; + if (getPortOperSpeed(port, speed)) { - SWSS_LOG_NOTICE("Got port state change for port id 0x%" PRIx64 " which does not exist, possibly outdated event", id); - continue; + SWSS_LOG_NOTICE("%s oper speed is %d", port.m_alias.c_str(), speed); + updateDbPortOperSpeed(port, speed); } - - updatePortOperStatus(port, status); - if (status == SAI_PORT_OPER_STATUS_UP) + else { - sai_uint32_t speed; - if (getPortOperSpeed(port, speed)) - { - SWSS_LOG_NOTICE("%s oper speed is %d", port.m_alias.c_str(), speed); - updateDbPortOperSpeed(port, speed); - } - else - { - updateDbPortOperSpeed(port, 0); - } - sai_port_fec_mode_t fec_mode; - string fec_str; - if (oper_fec_sup && getPortOperFec(port, fec_mode)) - { - if (!m_portHlpr.fecToStr(fec_str, fec_mode)) - { - SWSS_LOG_ERROR("Error unknown fec mode %d while querying port %s fec mode", - static_cast(fec_mode), port.m_alias.c_str()); - fec_str = "N/A"; - } - updateDbPortOperFec(port,fec_str); - } - else - { - updateDbPortOperFec(port, "N/A"); - } - } else { - if (port_oper_err) + updateDbPortOperSpeed(port, 0); + } + sai_port_fec_mode_t fec_mode; + string fec_str; + if (oper_fec_sup && getPortOperFec(port, fec_mode)) + { + if (!m_portHlpr.fecToStr(fec_str, fec_mode)) { - updatePortErrorStatus(port, port_oper_err); + SWSS_LOG_ERROR("Error unknown fec mode %d while querying port %s fec mode", + static_cast(fec_mode), port.m_alias.c_str()); + fec_str = "N/A"; } + updateDbPortOperFec(port,fec_str); + } + else + { + updateDbPortOperFec(port, "N/A"); + } + } else { + if (port_oper_err) + { + updatePortErrorStatus(port, port_oper_err); } - - /* update m_portList */ - m_portList[port.m_alias] = port; } - sai_deserialize_free_port_oper_status_ntf(count, portoperstatus); + /* update m_portList */ + m_portList[port.m_alias] = port; } - else if (&consumer == m_portHostTxReadyNotificationConsumer && op == "port_host_tx_ready") - { - sai_object_id_t port_id; - sai_object_id_t switch_id; - sai_port_host_tx_ready_status_t host_tx_ready_status; - sai_deserialize_port_host_tx_ready_ntf(data, switch_id, port_id, host_tx_ready_status); - SWSS_LOG_DEBUG("Recieved host_tx_ready notification for port 0x%" PRIx64, port_id); + sai_deserialize_free_port_oper_status_ntf(count, portoperstatus); +} - Port p; - if (!getPort(port_id, p)) - { - SWSS_LOG_ERROR("Failed to get port object for port id 0x%" PRIx64, port_id); - return; - } - setHostTxReady(p, host_tx_ready_status == SAI_PORT_HOST_TX_READY_STATUS_READY ? "true" : "false"); +void PortsOrch::handlePortHostTxReadyNotification(const std::string &data) +{ + sai_object_id_t port_id; + sai_object_id_t switch_id; + sai_port_host_tx_ready_status_t host_tx_ready_status; + + sai_deserialize_port_host_tx_ready_ntf(data, switch_id, port_id, host_tx_ready_status); + SWSS_LOG_DEBUG("Recieved host_tx_ready notification for port 0x%" PRIx64, port_id); + + Port p; + if (!getPort(port_id, p)) + { + SWSS_LOG_ERROR("Failed to get port object for port id 0x%" PRIx64, port_id); + return; } + setHostTxReady(p, host_tx_ready_status == SAI_PORT_HOST_TX_READY_STATUS_READY ? "true" : "false"); } void PortsOrch::updatePortErrorStatus(Port &port, sai_port_error_status_t errstatus) diff --git a/orchagent/portsorch.h b/orchagent/portsorch.h index e0d392547c3..22bcf9c1513 100644 --- a/orchagent/portsorch.h +++ b/orchagent/portsorch.h @@ -435,7 +435,9 @@ class PortsOrch : public Orch, public Subject void doTransceiverPresenceCheck(Consumer &consumer); void doTask(NotificationConsumer &consumer); - void handleNotification(NotificationConsumer &consumer, KeyOpFieldsValuesTuple& entry); + void handleNotification(KeyOpFieldsValuesTuple& entry); + void handlePortStateChangeNotification(const std::string &data); + void handlePortHostTxReadyNotification(const std::string &data); void doTask(swss::SelectableTimer &timer); void removePortFromLanesMap(string alias); diff --git a/orchagent/sainotificationorch.cpp b/orchagent/sainotificationorch.cpp new file mode 100644 index 00000000000..023e26458d3 --- /dev/null +++ b/orchagent/sainotificationorch.cpp @@ -0,0 +1,24 @@ +#include "sainotificationorch.h" + +#include "logger.h" + +SaiNotificationOrch *gSaiNotificationOrch = nullptr; + +SaiNotificationOrch::SaiNotificationOrch() + : Orch() +{ + SWSS_LOG_ENTER(); + + m_queue = getSaiNotificationQueue(); + m_dispatcher = getSaiNotificationDispatcher(); + + Orch::addExecutor(createSaiNotificationQueueExecutor( + m_queue, this, m_dispatcher, "SAI_NOTIFICATION_QUEUE")); +} + +void SaiNotificationOrch::registerHandler(const std::string &op, + SaiNotificationDispatcher::Handler handler, + SaiNotificationDispatcher::ReadinessPredicate ready) +{ + m_dispatcher->registerHandler(op, std::move(handler), std::move(ready)); +} diff --git a/orchagent/sainotificationorch.h b/orchagent/sainotificationorch.h new file mode 100644 index 00000000000..c5494afad90 --- /dev/null +++ b/orchagent/sainotificationorch.h @@ -0,0 +1,38 @@ +#pragma once + +#include "orch.h" +#include "notifications.h" + +#include +#include + +/* + * SaiNotificationOrch + * + * Drains the in-process SaiNotificationQueue on the orchagent main loop and + * dispatches entries to handlers registered by feature orchs. + * + * Lifecycle: + * + * - OrchDaemon constructs this orch once in ZMQ mode, before any orch that + * registers an in-process SAI notification handler. + * - Each Orch that handles a queued notification calls + * gSaiNotificationOrch->registerHandler(op, handler, readiness) + * from its constructor. Registration is a no-op when the global pointer + * is null (non-ZMQ mode). + */ +class SaiNotificationOrch : public Orch +{ +public: + SaiNotificationOrch(); + + void registerHandler(const std::string &op, + SaiNotificationDispatcher::Handler handler, + SaiNotificationDispatcher::ReadinessPredicate ready = nullptr); + +private: + SaiNotificationQueue *m_queue; + SaiNotificationDispatcher *m_dispatcher; +}; + +extern SaiNotificationOrch *gSaiNotificationOrch; diff --git a/orchagent/twamporch.cpp b/orchagent/twamporch.cpp index 834d13fe060..f582b9d3ce8 100644 --- a/orchagent/twamporch.cpp +++ b/orchagent/twamporch.cpp @@ -9,6 +9,7 @@ #include "tokenize.h" #include "notifier.h" #include "notifications.h" +#include "sainotificationorch.h" #include @@ -156,6 +157,17 @@ TwampOrch::TwampOrch(TableConnector confDbConnector, TableConnector stateDbConne Orch::addExecutor(twampNotifier); register_event_notif = false; + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_TWAMP_SESSION_EVENT, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }, + [this]() { return m_portsOrch->allPortsReady(); }); + } + /* Initialize DB connectors */ m_asicDb = shared_ptr(new DBConnector("ASIC_DB", 0)); m_countersDb = shared_ptr(new DBConnector("COUNTERS_DB", 0)); @@ -994,65 +1006,76 @@ void TwampOrch::doTask(NotificationConsumer& consumer) return; } - if (op == "twamp_session_event") + KeyOpFieldsValuesTuple entry = std::make_tuple(data, op, values); + handleNotification(entry); +} + +void TwampOrch::handleNotification(KeyOpFieldsValuesTuple &entry) +{ + if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_TWAMP_SESSION_EVENT) { - uint32_t count = 0; - sai_twamp_session_event_notification_data_t *twamp_session = nullptr; + handleTwampSessionEventNotification(kfvKey(entry)); + } +} - sai_deserialize_twamp_session_event_ntf(data, count, &twamp_session); +void TwampOrch::handleTwampSessionEventNotification(const std::string &data) +{ + uint32_t count = 0; + sai_twamp_session_event_notification_data_t *twamp_session = nullptr; + + sai_deserialize_twamp_session_event_ntf(data, count, &twamp_session); - for (uint32_t i = 0; i < count; i++) + for (uint32_t i = 0; i < count; i++) + { + string name; + sai_object_id_t session_id = twamp_session[i].twamp_session_id; + sai_twamp_session_state_t session_state = twamp_session[i].session_state; + uint32_t stats_index = twamp_session[i].session_stats.index; + + if (!getSessionName(session_id, name)) { - string name; - sai_object_id_t session_id = twamp_session[i].twamp_session_id; - sai_twamp_session_state_t session_state = twamp_session[i].session_state; - uint32_t stats_index = twamp_session[i].session_stats.index; + continue; + } - if (!getSessionName(session_id, name)) - { - continue; - } + /* update state db */ + if (session_state == SAI_TWAMP_SESSION_STATE_ACTIVE) + { + setSessionStatus(name, TWAMP_SESSION_STATUS_ACTIVE); + } + else + { + setSessionStatus(name, TWAMP_SESSION_STATUS_INACTIVE); + } - /* update state db */ - if (session_state == SAI_TWAMP_SESSION_STATE_ACTIVE) + /* save counter db */ + if (twamp_session[i].session_stats.number_of_counters) + { + if (0 == stats_index) { - setSessionStatus(name, TWAMP_SESSION_STATUS_ACTIVE); + continue; } - else + else if (1 == stats_index) { - setSessionStatus(name, TWAMP_SESSION_STATUS_INACTIVE); + addCounterNameMap(name, session_id); } - /* save counter db */ - if (twamp_session[i].session_stats.number_of_counters) + vector hw_stats; + hw_stats.resize(twamp_session_stat_ids.size()); + for (uint32_t j = 0; j < twamp_session[i].session_stats.number_of_counters; j++) { - if (0 == stats_index) - { - continue; - } - else if (1 == stats_index) + uint32_t counters_id = twamp_session[i].session_stats.counters_ids[j]; + auto it = find(twamp_session_stat_ids.begin(), twamp_session_stat_ids.end(), counters_id); + if (it != twamp_session_stat_ids.end()) { - addCounterNameMap(name, session_id); + hw_stats[counters_id] = twamp_session[i].session_stats.counters[j]; } - - vector hw_stats; - hw_stats.resize(twamp_session_stat_ids.size()); - for (uint32_t j = 0; j < twamp_session[i].session_stats.number_of_counters; j++) - { - uint32_t counters_id = twamp_session[i].session_stats.counters_ids[j]; - auto it = find(twamp_session_stat_ids.begin(), twamp_session_stat_ids.end(), counters_id); - if (it != twamp_session_stat_ids.end()) - { - hw_stats[counters_id] = twamp_session[i].session_stats.counters[j]; - } - } - - saveSessionStatsLatest(session_id, stats_index, hw_stats); - calculateCounters(name, stats_index, hw_stats); - saveCountersTotal(name, session_id); } - } - sai_deserialize_free_twamp_session_event_ntf(count, twamp_session); + saveSessionStatsLatest(session_id, stats_index, hw_stats); + calculateCounters(name, stats_index, hw_stats); + saveCountersTotal(name, session_id); + } } + + sai_deserialize_free_twamp_session_event_ntf(count, twamp_session); } diff --git a/orchagent/twamporch.h b/orchagent/twamporch.h index 09134f6be4a..30c9cdd6118 100644 --- a/orchagent/twamporch.h +++ b/orchagent/twamporch.h @@ -130,6 +130,8 @@ class TwampOrch : public Orch void saveSessionStatsLatest(const sai_object_id_t session_id, const uint32_t index, const vector& stats); void calculateCounters(const string&, const uint32_t index, const vector& stats); void saveCountersTotal(const string&, const sai_object_id_t session_id); + void handleNotification(swss::KeyOpFieldsValuesTuple &entry); + void handleTwampSessionEventNotification(const std::string &data); void doTask(NotificationConsumer& consumer); }; diff --git a/tests/mock_tests/Makefile.am b/tests/mock_tests/Makefile.am index 157416d96c6..f2064566d6f 100644 --- a/tests/mock_tests/Makefile.am +++ b/tests/mock_tests/Makefile.am @@ -113,6 +113,7 @@ tests_SOURCES = aclorch_ut.cpp \ $(top_srcdir)/orchagent/orchdaemon.cpp \ $(top_srcdir)/orchagent/orch.cpp \ $(top_srcdir)/orchagent/notifications.cpp \ + $(top_srcdir)/orchagent/sainotificationorch.cpp \ $(top_srcdir)/orchagent/routeorch.cpp \ $(top_srcdir)/orchagent/mplsrouteorch.cpp \ $(top_srcdir)/orchagent/fgnhgorch.cpp \ diff --git a/tests/mock_tests/notifications_ut.cpp b/tests/mock_tests/notifications_ut.cpp index 5320b3a5143..b3f4ded8d88 100644 --- a/tests/mock_tests/notifications_ut.cpp +++ b/tests/mock_tests/notifications_ut.cpp @@ -1,231 +1,602 @@ #include "gtest/gtest.h" -#include "sairedis.h" + +#include +#include + #include "notifications.h" +#include "orch.h" +#include "sai_serialize.h" +#include "saiextensions.h" +#include "sairedis.h" +#include "swss/table.h" + +struct IcmpSaiSessionHandler +{ + static void on_state_change(uint32_t count, sai_icmp_echo_session_state_notification_t *data); +}; extern sai_redis_communication_mode_t gRedisCommunicationMode; +extern sai_object_id_t gSwitchId; +extern volatile sig_atomic_t gOrchShutdownRequested; -namespace notifications_zmq_test +namespace notifications_test { - using namespace std; - class NotificationsZmqForwardingTest : public ::testing::Test +using namespace std; +using namespace swss; + +class TestNotificationOrch : public Orch +{ +public: + TestNotificationOrch() + : Orch() { - protected: - sai_redis_communication_mode_t m_oldMode; + } +}; - void SetUp() override - { - m_oldMode = gRedisCommunicationMode; - gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC; - } +static void drainSaiNotificationQueue() +{ + auto queue = getSaiNotificationQueue(); + std::deque entries; + while (queue->hasData()) + { + queue->pops(entries); + entries.clear(); + } +} - void TearDown() override - { - gRedisCommunicationMode = m_oldMode; - } - }; +class SaiNotificationZmqTest : public ::testing::Test +{ +protected: + sai_redis_communication_mode_t m_oldMode; - TEST_F(NotificationsZmqForwardingTest, FdbEventForwarding) + void SetUp() override { - sai_fdb_event_notification_data_t data; - memset(&data, 0, sizeof(data)); - data.event_type = SAI_FDB_EVENT_LEARNED; - data.fdb_entry.switch_id = 0x1; - data.fdb_entry.bv_id = 0x2000; + m_oldMode = gRedisCommunicationMode; + gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC; + drainSaiNotificationQueue(); + } - // Set a MAC address - uint8_t mac[] = {0x00, 0x11, 0x22, 0x33, 0x44, 0x55}; - memcpy(data.fdb_entry.mac_address, mac, sizeof(mac)); + void TearDown() override + { + drainSaiNotificationQueue(); + gRedisCommunicationMode = m_oldMode; + } +}; - data.attr_count = 0; - data.attr = nullptr; +TEST_F(SaiNotificationZmqTest, EnqueueDroppedDuringShutdown) +{ + gOrchShutdownRequested = SIGTERM; - // Should not crash — Redis forwarding via mock hiredis (no-op) - ASSERT_NO_THROW(on_fdb_event(1, &data)); - } + enqueueSaiNotification("fdb_event", "data", std::vector()); - TEST_F(NotificationsZmqForwardingTest, FdbEventMultipleEntries) - { - const int count = 3; - sai_fdb_event_notification_data_t data[count]; - memset(data, 0, sizeof(data)); + EXPECT_EQ(getSaiNotificationQueue()->size(), 0u); - for (int i = 0; i < count; i++) - { - data[i].event_type = SAI_FDB_EVENT_LEARNED; - data[i].fdb_entry.switch_id = 0x1; - data[i].fdb_entry.bv_id = 0x2000 + i; - uint8_t mac[] = {0x00, 0x11, 0x22, 0x33, 0x44, (uint8_t)(0x50 + i)}; - memcpy(data[i].fdb_entry.mac_address, mac, sizeof(mac)); - data[i].attr_count = 0; - data[i].attr = nullptr; - } - - ASSERT_NO_THROW(on_fdb_event(count, data)); - } + gOrchShutdownRequested = 0; +} - TEST_F(NotificationsZmqForwardingTest, PortStateChangeForwarding) - { - sai_port_oper_status_notification_t data; - memset(&data, 0, sizeof(data)); - data.port_id = 0x100; - data.port_state = SAI_PORT_OPER_STATUS_UP; +TEST(SaiNotificationQueueTest, EnqueueAndPop) +{ + SaiNotificationQueue queue(100, 2); + std::vector values; - ASSERT_NO_THROW(on_port_state_change(1, &data)); - } + ASSERT_FALSE(queue.hasData()); - TEST_F(NotificationsZmqForwardingTest, PortStateChangeMultiple) - { - const int count = 4; - sai_port_oper_status_notification_t data[count]; - memset(data, 0, sizeof(data)); + queue.enqueue("port_state_change", "data1", values); + queue.enqueue("port_state_change", "data2", values); + queue.enqueue("port_state_change", "data3", values); - for (int i = 0; i < count; i++) - { - data[i].port_id = 0x100 + i; - data[i].port_state = (i % 2 == 0) ? SAI_PORT_OPER_STATUS_UP : SAI_PORT_OPER_STATUS_DOWN; - } + ASSERT_TRUE(queue.hasData()); + ASSERT_TRUE(queue.hasCachedData()); - ASSERT_NO_THROW(on_port_state_change(count, data)); - } + std::deque entries; + queue.pops(entries); - TEST_F(NotificationsZmqForwardingTest, BfdSessionStateChangeForwarding) - { - sai_bfd_session_state_notification_t data; - memset(&data, 0, sizeof(data)); - data.bfd_session_id = 0x200; - data.session_state = SAI_BFD_SESSION_STATE_UP; + ASSERT_EQ(entries.size(), static_cast(2)); + EXPECT_EQ(kfvOp(entries[0]), "port_state_change"); + EXPECT_EQ(kfvKey(entries[0]), "data1"); + EXPECT_EQ(kfvOp(entries[1]), "port_state_change"); + EXPECT_EQ(kfvKey(entries[1]), "data2"); + ASSERT_TRUE(queue.hasData()); - ASSERT_NO_THROW(on_bfd_session_state_change(1, &data)); - } + queue.pops(entries); - TEST_F(NotificationsZmqForwardingTest, BfdSessionStateChangeMultiple) - { - const int count = 2; - sai_bfd_session_state_notification_t data[count]; - memset(data, 0, sizeof(data)); + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), "port_state_change"); + EXPECT_EQ(kfvKey(entries[0]), "data3"); + ASSERT_FALSE(queue.hasData()); +} - data[0].bfd_session_id = 0x200; - data[0].session_state = SAI_BFD_SESSION_STATE_UP; - data[1].bfd_session_id = 0x201; - data[1].session_state = SAI_BFD_SESSION_STATE_DOWN; +TEST(SaiNotificationQueueTest, DispatcherInvokesRegisteredHandler) +{ + SaiNotificationDispatcher dispatcher; + std::vector values; + KeyOpFieldsValuesTuple entry("payload", "port_state_change", values); + bool called = false; + std::string payload; + + dispatcher.registerHandler( + "port_state_change", + [&](KeyOpFieldsValuesTuple &dispatchedEntry) + { + called = true; + payload = kfvKey(dispatchedEntry); + }); - ASSERT_NO_THROW(on_bfd_session_state_change(count, data)); - } + dispatcher.dispatch(entry); - TEST_F(NotificationsZmqForwardingTest, TwampSessionEventForwarding) - { - sai_twamp_session_event_notification_data_t data; - memset(&data, 0, sizeof(data)); - data.twamp_session_id = 0x300; - data.session_state = SAI_TWAMP_SESSION_STATE_ACTIVE; + EXPECT_TRUE(called); + EXPECT_EQ(payload, "payload"); +} - ASSERT_NO_THROW(on_twamp_session_event(1, &data)); - } +TEST_F(SaiNotificationZmqTest, PortStateChangeCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); - TEST_F(NotificationsZmqForwardingTest, PortHostTxReadyForwarding) - { - sai_object_id_t switch_id = 0x1; - sai_object_id_t port_id = 0x400; - sai_port_host_tx_ready_status_t status = SAI_PORT_HOST_TX_READY_STATUS_READY; + sai_port_oper_status_notification_t port_oper_status; + memset(&port_oper_status, 0, sizeof(port_oper_status)); + port_oper_status.port_id = 0x1000000000019; + port_oper_status.port_state = SAI_PORT_OPER_STATUS_UP; + port_oper_status.port_error_status = SAI_PORT_ERROR_STATUS_CLEAR; - ASSERT_NO_THROW(on_port_host_tx_ready(switch_id, port_id, status)); - } + on_port_state_change(1, &port_oper_status); - TEST_F(NotificationsZmqForwardingTest, PortHostTxReadyNotReady) - { - sai_object_id_t switch_id = 0x1; - sai_object_id_t port_id = 0x401; - sai_port_host_tx_ready_status_t status = SAI_PORT_HOST_TX_READY_STATUS_NOT_READY; + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); - ASSERT_NO_THROW(on_port_host_tx_ready(switch_id, port_id, status)); - } + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), "port_state_change"); - TEST_F(NotificationsZmqForwardingTest, TamTelTypeConfigChangeForwarding) - { - sai_object_id_t tam_tel_id = 0x500; + uint32_t count = 0; + sai_port_oper_status_notification_t *deserialized_status = nullptr; + sai_deserialize_port_oper_status_ntf(kfvKey(entries[0]), count, &deserialized_status); - ASSERT_NO_THROW(on_tam_tel_type_config_change(tam_tel_id)); - } + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized_status[0].port_id, port_oper_status.port_id); + EXPECT_EQ(deserialized_status[0].port_state, port_oper_status.port_state); + EXPECT_EQ(deserialized_status[0].port_error_status, port_oper_status.port_error_status); - // Verify callbacks are no-ops in non-ZMQ mode (no forwarding) - TEST(NotificationsNonZmqTest, FdbEventNoForwardingInRedisMode) - { - sai_redis_communication_mode_t oldMode = gRedisCommunicationMode; - gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC; + sai_deserialize_free_port_oper_status_ntf(count, deserialized_status); + ASSERT_FALSE(queue->hasData()); +} - sai_fdb_event_notification_data_t data; - memset(&data, 0, sizeof(data)); - data.event_type = SAI_FDB_EVENT_LEARNED; - data.fdb_entry.switch_id = 0x1; - data.attr_count = 0; - data.attr = nullptr; +TEST(SaiNotificationQueueTest, PeekFrontOp) +{ + SaiNotificationQueue queue(100, 1); + std::vector values; + std::string op; - // In Redis mode, callback should be a no-op (no forwarding needed) - ASSERT_NO_THROW(on_fdb_event(1, &data)); + ASSERT_FALSE(queue.peekFrontOp(op)); - gRedisCommunicationMode = oldMode; - } + queue.enqueue("port_state_change", "data1", values); + queue.enqueue("fdb_event", "data2", values); - TEST(NotificationsNonZmqTest, BfdNoForwardingInRedisMode) - { - sai_redis_communication_mode_t oldMode = gRedisCommunicationMode; - gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC; + ASSERT_TRUE(queue.peekFrontOp(op)); + EXPECT_EQ(op, "port_state_change"); - sai_bfd_session_state_notification_t data; - memset(&data, 0, sizeof(data)); - data.bfd_session_id = 0x200; - data.session_state = SAI_BFD_SESSION_STATE_UP; + std::deque entries; + queue.pops(entries); + ASSERT_EQ(entries.size(), static_cast(1)); - ASSERT_NO_THROW(on_bfd_session_state_change(1, &data)); + ASSERT_TRUE(queue.peekFrontOp(op)); + EXPECT_EQ(op, "fdb_event"); +} - gRedisCommunicationMode = oldMode; - } +TEST(SaiNotificationQueueTest, ReadinessPredicateReportsNotReady) +{ + SaiNotificationDispatcher dispatcher; - TEST(NotificationsNonZmqTest, PortStateChangeNoForwardingInRedisMode) - { - sai_redis_communication_mode_t oldMode = gRedisCommunicationMode; - gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC; + dispatcher.registerHandler( + "port_state_change", + [](KeyOpFieldsValuesTuple &) + { + }, + []() { return false; }); - sai_port_oper_status_notification_t data; - memset(&data, 0, sizeof(data)); - data.port_id = 0x100; - data.port_state = SAI_PORT_OPER_STATUS_UP; + EXPECT_FALSE(dispatcher.isReady("port_state_change")); - ASSERT_NO_THROW(on_port_state_change(1, &data)); + dispatcher.registerHandler( + "port_state_change", + [](KeyOpFieldsValuesTuple &) + { + }, + []() { return true; }); - gRedisCommunicationMode = oldMode; - } + EXPECT_TRUE(dispatcher.isReady("port_state_change")); +} - // Repeated calls to verify static thread_local connections stay stable - TEST_F(NotificationsZmqForwardingTest, RepeatedFdbEventsStable) - { - for (int i = 0; i < 100; i++) +TEST(SaiNotificationQueueExecutorTest, SaiNotificationQueueExecutor) +{ + SaiNotificationQueue queue(100, 10); + SaiNotificationDispatcher dispatcher; + TestNotificationOrch orch; + std::vector values; + bool called = false; + + dispatcher.registerHandler( + "bfd_session_state_change", + [&](KeyOpFieldsValuesTuple &) { - sai_fdb_event_notification_data_t data; - memset(&data, 0, sizeof(data)); - data.event_type = (i % 2 == 0) ? SAI_FDB_EVENT_LEARNED : SAI_FDB_EVENT_AGED; - data.fdb_entry.switch_id = 0x1; - data.fdb_entry.bv_id = 0x2000; - uint8_t mac[] = {0x00, 0x11, 0x22, 0x33, (uint8_t)(i >> 8), (uint8_t)(i & 0xff)}; - memcpy(data.fdb_entry.mac_address, mac, sizeof(mac)); - data.attr_count = 0; - data.attr = nullptr; - - ASSERT_NO_THROW(on_fdb_event(1, &data)); - } - } + called = true; + }); - TEST_F(NotificationsZmqForwardingTest, RepeatedBfdEventsStable) - { - for (int i = 0; i < 50; i++) + queue.enqueue("bfd_session_state_change", "bfd", values); + + std::unique_ptr executor( + createSaiNotificationQueueExecutor(&queue, &orch, &dispatcher, "TEST_EXECUTOR")); + + ASSERT_TRUE(queue.hasData()); + executor->execute(); + + EXPECT_TRUE(called); + EXPECT_FALSE(queue.hasData()); +} + +TEST(SaiNotificationQueueExecutorTest, SaiNotificationQueueExecutorHeadOfLineBlocksUntilReady) +{ + SaiNotificationQueue queue(100, 10); + SaiNotificationDispatcher dispatcher; + TestNotificationOrch orch; + std::vector values; + bool portsReady = false; + int fdbCalls = 0; + int bfdCalls = 0; + std::deque dispatchOrder; + + dispatcher.registerHandler( + "fdb_event", + [&](KeyOpFieldsValuesTuple &) + { + fdbCalls++; + dispatchOrder.push_back("fdb_event"); + }, + [&]() { return portsReady; }); + + dispatcher.registerHandler( + "bfd_session_state_change", + [&](KeyOpFieldsValuesTuple &) { - sai_bfd_session_state_notification_t data; - memset(&data, 0, sizeof(data)); - data.bfd_session_id = 0x200 + i; - data.session_state = (i % 2 == 0) ? SAI_BFD_SESSION_STATE_UP : SAI_BFD_SESSION_STATE_DOWN; + bfdCalls++; + dispatchOrder.push_back("bfd_session_state_change"); + }); - ASSERT_NO_THROW(on_bfd_session_state_change(1, &data)); - } - } + queue.enqueue("fdb_event", "fdb", values); + queue.enqueue("bfd_session_state_change", "bfd", values); + + std::unique_ptr executor( + createSaiNotificationQueueExecutor(&queue, &orch, &dispatcher, "TEST_EXECUTOR")); + + executor->execute(); + + EXPECT_EQ(fdbCalls, 0); + EXPECT_EQ(bfdCalls, 0); + EXPECT_EQ(queue.size(), 2u); + EXPECT_TRUE(dispatchOrder.empty()); + + portsReady = true; + executor->execute(); + + EXPECT_EQ(fdbCalls, 1); + EXPECT_EQ(bfdCalls, 1); + EXPECT_FALSE(queue.hasData()); + ASSERT_EQ(dispatchOrder.size(), 2u); + EXPECT_EQ(dispatchOrder[0], "fdb_event"); + EXPECT_EQ(dispatchOrder[1], "bfd_session_state_change"); +} + +TEST(SaiNotificationQueueTest, MissingHandlerLogsWarning) +{ + SaiNotificationDispatcher dispatcher; + std::vector values; + KeyOpFieldsValuesTuple entry("payload", "unknown_op", values); + + dispatcher.dispatch(entry); + SUCCEED(); +} + +TEST_F(SaiNotificationZmqTest, PortHostTxReadyCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + on_port_host_tx_ready(0x1000000000001, 0x1000000000019, SAI_PORT_HOST_TX_READY_STATUS_READY); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), "port_host_tx_ready"); + + sai_object_id_t switch_id = 0; + sai_object_id_t port_id = 0; + sai_port_host_tx_ready_status_t status = SAI_PORT_HOST_TX_READY_STATUS_NOT_READY; + sai_deserialize_port_host_tx_ready_ntf(kfvKey(entries[0]), switch_id, port_id, status); + + EXPECT_EQ(switch_id, static_cast(0x1000000000001)); + EXPECT_EQ(port_id, static_cast(0x1000000000019)); + EXPECT_EQ(status, SAI_PORT_HOST_TX_READY_STATUS_READY); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, BfdSessionStateChangeCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_bfd_session_state_notification_t bfd_session_state; + memset(&bfd_session_state, 0, sizeof(bfd_session_state)); + bfd_session_state.bfd_session_id = 0x1000000000020; + bfd_session_state.session_state = SAI_BFD_SESSION_STATE_UP; + + on_bfd_session_state_change(1, &bfd_session_state); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), "bfd_session_state_change"); + + uint32_t count = 0; + sai_bfd_session_state_notification_t *deserialized = nullptr; + sai_deserialize_bfd_session_state_ntf(kfvKey(entries[0]), count, &deserialized); + + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized[0].bfd_session_id, bfd_session_state.bfd_session_id); + EXPECT_EQ(deserialized[0].session_state, bfd_session_state.session_state); + + sai_deserialize_free_bfd_session_state_ntf(count, deserialized); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, IcmpEchoSessionStateChangeCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_icmp_echo_session_state_notification_t icmp_session_state; + memset(&icmp_session_state, 0, sizeof(icmp_session_state)); + icmp_session_state.icmp_echo_session_id = 0x1000000000021; + icmp_session_state.session_state = SAI_ICMP_ECHO_SESSION_STATE_UP; + + IcmpSaiSessionHandler::on_state_change(1, &icmp_session_state); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), "icmp_echo_session_state_change"); + + uint32_t count = 0; + sai_icmp_echo_session_state_notification_t *deserialized = nullptr; + sai_deserialize_icmp_echo_session_state_ntf(kfvKey(entries[0]), count, &deserialized); + + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized[0].icmp_echo_session_id, icmp_session_state.icmp_echo_session_id); + EXPECT_EQ(deserialized[0].session_state, icmp_session_state.session_state); + + sai_deserialize_free_icmp_echo_session_state_ntf(count, deserialized); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, TwampSessionEventCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_twamp_session_event_notification_data_t twamp_session; + memset(&twamp_session, 0, sizeof(twamp_session)); + twamp_session.twamp_session_id = 0x1000000000022; + twamp_session.session_state = SAI_TWAMP_SESSION_STATE_ACTIVE; + + on_twamp_session_event(1, &twamp_session); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), "twamp_session_event"); + + uint32_t count = 0; + sai_twamp_session_event_notification_data_t *deserialized = nullptr; + sai_deserialize_twamp_session_event_ntf(kfvKey(entries[0]), count, &deserialized); + + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized[0].twamp_session_id, twamp_session.twamp_session_id); + EXPECT_EQ(deserialized[0].session_state, twamp_session.session_state); + + sai_deserialize_free_twamp_session_event_ntf(count, deserialized); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, HaSetEventCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_ha_set_event_data_t ha_set_event; + memset(&ha_set_event, 0, sizeof(ha_set_event)); + ha_set_event.ha_set_id = 0x1000000000030; + ha_set_event.event_type = SAI_HA_SET_EVENT_DP_CHANNEL_UP; + + on_ha_set_event(1, &ha_set_event); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), SAI_SWITCH_NOTIFICATION_NAME_HA_SET_EVENT); + + uint32_t count = 0; + sai_ha_set_event_data_t *deserialized = nullptr; + sai_deserialize_ha_set_event_ntf(kfvKey(entries[0]), count, &deserialized); + + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized[0].ha_set_id, ha_set_event.ha_set_id); + EXPECT_EQ(deserialized[0].event_type, ha_set_event.event_type); + + sai_deserialize_free_ha_set_event_ntf(count, deserialized); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, HaScopeEventCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_ha_scope_event_data_t ha_scope_event; + memset(&ha_scope_event, 0, sizeof(ha_scope_event)); + ha_scope_event.ha_scope_id = 0x1000000000031; + ha_scope_event.event_type = SAI_HA_SCOPE_EVENT_STATE_CHANGED; + ha_scope_event.ha_state = SAI_DASH_HA_STATE_ACTIVE; + + on_ha_scope_event(1, &ha_scope_event); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), SAI_SWITCH_NOTIFICATION_NAME_HA_SCOPE_EVENT); + + uint32_t count = 0; + sai_ha_scope_event_data_t *deserialized = nullptr; + sai_deserialize_ha_scope_event_ntf(kfvKey(entries[0]), count, &deserialized); + + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized[0].ha_scope_id, ha_scope_event.ha_scope_id); + EXPECT_EQ(deserialized[0].event_type, ha_scope_event.event_type); + + sai_deserialize_free_ha_scope_event_ntf(count, deserialized); + ASSERT_FALSE(queue->hasData()); } + +TEST_F(SaiNotificationZmqTest, FlowBulkGetSessionEventCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_flow_bulk_get_session_event_data_t flow_event; + memset(&flow_event, 0, sizeof(flow_event)); + flow_event.event_type = SAI_FLOW_BULK_GET_SESSION_EVENT_FINISHED; + + on_flow_bulk_get_session_event(0x1000000000032, 1, &flow_event); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT); + + sai_object_id_t session_id = SAI_NULL_OBJECT_ID; + uint32_t count = 0; + sai_flow_bulk_get_session_event_data_t *deserialized = nullptr; + sai_deserialize_flow_bulk_get_session_event_ntf(kfvKey(entries[0]), session_id, count, &deserialized); + + EXPECT_EQ(session_id, static_cast(0x1000000000032)); + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized[0].event_type, flow_event.event_type); + + sai_deserialize_free_flow_bulk_get_session_event_ntf(count, deserialized); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, SwitchMacsecPostStatusCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + on_switch_macsec_post_status_notify(gSwitchId, SAI_SWITCH_MACSEC_POST_STATUS_PASS); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), SAI_SWITCH_NOTIFICATION_NAME_SWITCH_MACSEC_POST_STATUS); + + sai_object_id_t switch_id = SAI_NULL_OBJECT_ID; + sai_switch_macsec_post_status_t status = SAI_SWITCH_MACSEC_POST_STATUS_UNKNOWN; + sai_deserialize_switch_macsec_post_status_ntf(kfvKey(entries[0]), switch_id, status); + + EXPECT_EQ(switch_id, gSwitchId); + EXPECT_EQ(status, SAI_SWITCH_MACSEC_POST_STATUS_PASS); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, MacsecPostStatusCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_object_id_t macsec_id = 0x1000000000033; + + on_macsec_post_status_notify(macsec_id, SAI_MACSEC_POST_STATUS_PASS); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), SAI_SWITCH_NOTIFICATION_NAME_MACSEC_POST_STATUS); + + sai_object_id_t deserialized_macsec_id = SAI_NULL_OBJECT_ID; + sai_macsec_post_status_t status = SAI_MACSEC_POST_STATUS_UNKNOWN; + sai_deserialize_macsec_post_status_ntf(kfvKey(entries[0]), deserialized_macsec_id, status); + + EXPECT_EQ(deserialized_macsec_id, macsec_id); + EXPECT_EQ(status, SAI_MACSEC_POST_STATUS_PASS); + ASSERT_FALSE(queue->hasData()); +} + +TEST_F(SaiNotificationZmqTest, TamTelTypeConfigChangeCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_object_id_t tam_tel_id = 0x1000000000034; + + on_tam_tel_type_config_change(tam_tel_id); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), SAI_SWITCH_NOTIFICATION_NAME_TAM_TEL_TYPE_CONFIG_CHANGE); + + sai_object_id_t deserialized_tam_tel_id = SAI_NULL_OBJECT_ID; + sai_deserialize_object_id(kfvKey(entries[0]), deserialized_tam_tel_id); + + EXPECT_EQ(deserialized_tam_tel_id, tam_tel_id); + ASSERT_FALSE(queue->hasData()); +} + +TEST(NotificationsNonZmqTest, PortStateChangeNoEnqueueInRedisMode) +{ + sai_redis_communication_mode_t oldMode = gRedisCommunicationMode; + gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC; + drainSaiNotificationQueue(); + + sai_port_oper_status_notification_t port_oper_status; + memset(&port_oper_status, 0, sizeof(port_oper_status)); + port_oper_status.port_id = 0x100; + port_oper_status.port_state = SAI_PORT_OPER_STATUS_UP; + + on_port_state_change(1, &port_oper_status); + + EXPECT_FALSE(getSaiNotificationQueue()->hasData()); + + gRedisCommunicationMode = oldMode; +} + +TEST(NotificationsNonZmqTest, TamTelTypeConfigChangeNoEnqueueInRedisMode) +{ + sai_redis_communication_mode_t oldMode = gRedisCommunicationMode; + gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC; + drainSaiNotificationQueue(); + + on_tam_tel_type_config_change(0x500); + + EXPECT_FALSE(getSaiNotificationQueue()->hasData()); + + gRedisCommunicationMode = oldMode; +} + +} // namespace notifications_test diff --git a/tests/mock_tests/portsorch_ut.cpp b/tests/mock_tests/portsorch_ut.cpp index cb39122f882..4aa17690bd7 100644 --- a/tests/mock_tests/portsorch_ut.cpp +++ b/tests/mock_tests/portsorch_ut.cpp @@ -13,7 +13,6 @@ #define private public #include "pfcactionhandler.h" #include "switchorch.h" -#include "notifications.h" #include #undef private #define private public @@ -948,12 +947,6 @@ namespace portsorch_test gPortsOrch->doTask(*consumer); mockReply = nullptr; - // Call the orchagent port state change callback method with zmq mode - sai_redis_communication_mode_t oldRedisCommunicationMode = gRedisCommunicationMode; - gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC; - on_port_state_change(1, &port_oper_status); - gRedisCommunicationMode = oldRedisCommunicationMode; - gPortsOrch->getPort("Ethernet0", port); ASSERT_TRUE(port.m_oper_status == oper_status); ASSERT_TRUE(port.m_flap_count == count+1); From 7646778b2bfc9c0ac00b458c041de831e022af14 Mon Sep 17 00:00:00 2001 From: Vijay Pandian Date: Thu, 30 Jul 2026 23:58:23 -0400 Subject: [PATCH 2/3] [orchagent]: Add FDB and non-ZMQ notification unit tests Signed-off-by: Vijay Pandian --- tests/mock_tests/notifications_ut.cpp | 75 +++++++++++++++++++++++++++ 1 file changed, 75 insertions(+) diff --git a/tests/mock_tests/notifications_ut.cpp b/tests/mock_tests/notifications_ut.cpp index b3f4ded8d88..6c243b5821a 100644 --- a/tests/mock_tests/notifications_ut.cpp +++ b/tests/mock_tests/notifications_ut.cpp @@ -161,6 +161,43 @@ TEST_F(SaiNotificationZmqTest, PortStateChangeCallbackEnqueuesInZmqMode) ASSERT_FALSE(queue->hasData()); } +TEST_F(SaiNotificationZmqTest, FdbEventCallbackEnqueuesInZmqMode) +{ + auto queue = getSaiNotificationQueue(); + + sai_fdb_event_notification_data_t fdb_data; + memset(&fdb_data, 0, sizeof(fdb_data)); + fdb_data.event_type = SAI_FDB_EVENT_LEARNED; + fdb_data.fdb_entry.switch_id = 0x21000000000000; + fdb_data.fdb_entry.bv_id = 0x26000000000a6c; + uint8_t mac[] = {0x52, 0x54, 0x00, 0x11, 0x22, 0x33}; + memcpy(fdb_data.fdb_entry.mac_address, mac, sizeof(mac)); + fdb_data.attr_count = 0; + fdb_data.attr = nullptr; + + on_fdb_event(1, &fdb_data); + + ASSERT_TRUE(queue->hasData()); + std::deque entries; + queue->pops(entries); + + ASSERT_EQ(entries.size(), static_cast(1)); + EXPECT_EQ(kfvOp(entries[0]), "fdb_event"); + + uint32_t count = 0; + sai_fdb_event_notification_data_t *deserialized = nullptr; + sai_deserialize_fdb_event_ntf(kfvKey(entries[0]), count, &deserialized); + + ASSERT_EQ(count, static_cast(1)); + EXPECT_EQ(deserialized[0].event_type, fdb_data.event_type); + EXPECT_EQ(deserialized[0].fdb_entry.switch_id, fdb_data.fdb_entry.switch_id); + EXPECT_EQ(deserialized[0].fdb_entry.bv_id, fdb_data.fdb_entry.bv_id); + EXPECT_EQ(memcmp(deserialized[0].fdb_entry.mac_address, mac, sizeof(mac)), 0); + + sai_deserialize_free_fdb_event_ntf(count, deserialized); + ASSERT_FALSE(queue->hasData()); +} + TEST(SaiNotificationQueueTest, PeekFrontOp) { SaiNotificationQueue queue(100, 1); @@ -599,4 +636,42 @@ TEST(NotificationsNonZmqTest, TamTelTypeConfigChangeNoEnqueueInRedisMode) gRedisCommunicationMode = oldMode; } +TEST(NotificationsNonZmqTest, FdbEventNoEnqueueInRedisMode) +{ + sai_redis_communication_mode_t oldMode = gRedisCommunicationMode; + gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC; + drainSaiNotificationQueue(); + + sai_fdb_event_notification_data_t fdb_data; + memset(&fdb_data, 0, sizeof(fdb_data)); + fdb_data.event_type = SAI_FDB_EVENT_LEARNED; + fdb_data.fdb_entry.switch_id = 0x1; + fdb_data.attr_count = 0; + fdb_data.attr = nullptr; + + on_fdb_event(1, &fdb_data); + + EXPECT_FALSE(getSaiNotificationQueue()->hasData()); + + gRedisCommunicationMode = oldMode; +} + +TEST(NotificationsNonZmqTest, BfdSessionStateChangeNoEnqueueInRedisMode) +{ + sai_redis_communication_mode_t oldMode = gRedisCommunicationMode; + gRedisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC; + drainSaiNotificationQueue(); + + sai_bfd_session_state_notification_t bfd_session_state; + memset(&bfd_session_state, 0, sizeof(bfd_session_state)); + bfd_session_state.bfd_session_id = 0x200; + bfd_session_state.session_state = SAI_BFD_SESSION_STATE_UP; + + on_bfd_session_state_change(1, &bfd_session_state); + + EXPECT_FALSE(getSaiNotificationQueue()->hasData()); + + gRedisCommunicationMode = oldMode; +} + } // namespace notifications_test From 8d744a786b55c5ed66ea4cef234d827cfd4bcf50 Mon Sep 17 00:00:00 2001 From: Vijay Pandian Date: Fri, 31 Jul 2026 00:36:25 -0400 Subject: [PATCH 3/3] [orchagent]: Fix duplicate icmp notifier executor registration Remove extra Orch::addExecutor() introduced during rebase conflict resolution in IcmpOrch constructor. Signed-off-by: Vijay Pandian --- orchagent/icmporch.cpp | 2 -- 1 file changed, 2 deletions(-) diff --git a/orchagent/icmporch.cpp b/orchagent/icmporch.cpp index 777f37352a5..43b5f154b76 100644 --- a/orchagent/icmporch.cpp +++ b/orchagent/icmporch.cpp @@ -74,8 +74,6 @@ IcmpOrch::IcmpOrch(DBConnector *db, string tableName, TableConnector stateDbIcmp auto icmpStateNotifier = new Notifier(m_icmpStateNotificationConsumer, this, "ICMP_STATE_NOTIFICATIONS"); Orch::addExecutor(icmpStateNotifier); - Orch::addExecutor(icmpStateNotifier); - if (gSaiNotificationOrch) { gSaiNotificationOrch->registerHandler(