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..43b5f154b76 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,16 @@ IcmpOrch::IcmpOrch(DBConnector *db, string tableName, TableConnector stateDbIcmp auto icmpStateNotifier = new Notifier(m_icmpStateNotificationConsumer, this, "ICMP_STATE_NOTIFICATIONS"); Orch::addExecutor(icmpStateNotifier); + if (gSaiNotificationOrch) + { + gSaiNotificationOrch->registerHandler( + SAI_SWITCH_NOTIFICATION_NAME_ICMP_ECHO_SESSION_STATE_CHANGE, + [this](KeyOpFieldsValuesTuple &entry) + { + handleNotification(entry); + }); + } + initializeCounters(); } @@ -200,47 +212,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); + + 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; - SWSS_LOG_INFO("Got ICMP session state change notification id:%" PRIx64 " state: %s", id, m_session_state_lkup.at(state).c_str()); + SWSS_LOG_INFO("Got ICMP session state change notification id:%" PRIx64 " state: %s", id, m_session_state_lkup.at(state).c_str()); - 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; - } + 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; + } - // 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); + // 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); - fvVector.push_back({IcmpSaiSessionHandler::m_state_fname, m_session_state_lkup.at(state)}); + fvVector.push_back({IcmpSaiSessionHandler::m_state_fname, m_session_state_lkup.at(state)}); - m_stateIcmpSessionTable.set(key, fvVector); + m_stateIcmpSessionTable.set(key, fvVector); - 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()); + 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_icmp_session_lookup[id].state = state; - m_icmp_session_lookup[id].init_state = false; - } + 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); } + + sai_deserialize_free_icmp_echo_session_state_ntf(count, icmpSessionState); } bool IcmpOrch::create_icmp_session(const string& key, const vector& data) @@ -734,14 +757,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..6c243b5821a 100644 --- a/tests/mock_tests/notifications_ut.cpp +++ b/tests/mock_tests/notifications_ut.cpp @@ -1,231 +1,677 @@ #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()); +} + +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()); +} - 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; +} + +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 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);