From 03bd6db875130089f39417cdf0dda2e5a894b145 Mon Sep 17 00:00:00 2001 From: Leonid Chernin Date: Mon, 15 Sep 2025 14:04:04 +0300 Subject: [PATCH] nvmeofgw: beacon diff implementation in the monitor unit tests works and tested new fields beacons coming according to new schema fixes for failover/failback/gw fast-reboot. src/messages/MNVMeofGwBeacon.h: add sequence number truncate prev_beacon_subsystems on beacon sequence mismatch do not process OOO map Signed-off-by: Alexander Indenbaum --- src/CMakeLists.txt | 1 + src/messages/MNVMeofGwBeacon.h | 35 ++++++-- src/mon/NVMeofGwSerialize.h | 77 ++++++++++++++---- src/mon/NVMeofGwTypes.h | 22 +++++- src/msg/Message.cc | 2 +- src/nvmeof/NVMeofGwMonitorClient.cc | 79 +++++++++++++++++-- src/nvmeof/NVMeofGwMonitorClient.h | 6 +- src/nvmeof/NVMeofGwUtils.cc | 53 +++++++++++++ src/nvmeof/NVMeofGwUtils.h | 23 ++++++ src/test/CMakeLists.txt | 8 ++ src/test/test_nvmeof_gw_utils.cc | 69 ++++++++++++++++ src/test/test_nvmeof_mon_encoding.cc | 114 ++++++++++++++++++++++----- 12 files changed, 438 insertions(+), 51 deletions(-) create mode 100644 src/nvmeof/NVMeofGwUtils.cc create mode 100644 src/nvmeof/NVMeofGwUtils.h create mode 100644 src/test/test_nvmeof_gw_utils.cc diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 3194054286b..00fae6aa5c6 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -1076,6 +1076,7 @@ if(WITH_NVMEOF_GATEWAY_MONITOR_CLIENT) ${nvmeof_monitor_grpc_hdrs} ceph_nvmeof_monitor_client.cc nvmeof/NVMeofGwClient.cc + nvmeof/NVMeofGwUtils.cc nvmeof/NVMeofGwMonitorGroupClient.cc nvmeof/NVMeofGwMonitorClient.cc) add_executable(ceph-nvmeof-monitor-client ${ceph_nvmeof_monitor_client_srcs}) diff --git a/src/messages/MNVMeofGwBeacon.h b/src/messages/MNVMeofGwBeacon.h index 26fc8dcf3ac..a26d6d09f07 100644 --- a/src/messages/MNVMeofGwBeacon.h +++ b/src/messages/MNVMeofGwBeacon.h @@ -24,8 +24,9 @@ class MNVMeofGwBeacon final : public PaxosServiceMessage { private: - static constexpr int HEAD_VERSION = 1; - static constexpr int COMPAT_VERSION = 1; + static constexpr int BEACON_VERSION_LEGACY = 1; // legacy beacon format (no diff support) + static constexpr int BEACON_VERSION_ENHANCED = 2; // enhanced beacon format (with diff support) + static constexpr int COMPAT_VERSION = BEACON_VERSION_LEGACY; // minimum version to decode enhanced format protected: std::string gw_id; @@ -35,10 +36,12 @@ protected: gw_availability_t availability; // in absence of beacon heartbeat messages it becomes inavailable epoch_t last_osd_epoch; epoch_t last_gwmap_epoch; + uint64_t sequence = 0; // sequence number for each beacon message public: MNVMeofGwBeacon() - : PaxosServiceMessage{MSG_MNVMEOF_GW_BEACON, 0, HEAD_VERSION, COMPAT_VERSION} + : PaxosServiceMessage{MSG_MNVMEOF_GW_BEACON, 0, BEACON_VERSION_ENHANCED, + COMPAT_VERSION}, sequence(0) { set_priority(CEPH_MSG_PRIO_HIGH); } @@ -49,11 +52,16 @@ public: const BeaconSubsystems& subsystems_, const gw_availability_t& availability_, const epoch_t& last_osd_epoch_, - const epoch_t& last_gwmap_epoch_ - ) - : PaxosServiceMessage{MSG_MNVMEOF_GW_BEACON, 0, HEAD_VERSION, COMPAT_VERSION}, + const epoch_t& last_gwmap_epoch_, + uint64_t sequence_ = 0, // default sequence for backward compatibility + bool enable_diff = false) // default to legacy behavior for backward compatibility + : PaxosServiceMessage{MSG_MNVMEOF_GW_BEACON, + static_cast(enable_diff ? 1 : 0), // user_version: 1=enhanced, 0=legacy + enable_diff ? BEACON_VERSION_ENHANCED : + BEACON_VERSION_LEGACY, COMPAT_VERSION},// Minimum compatible version gw_id(gw_id_), gw_pool(gw_pool_), gw_group(gw_group_), subsystems(subsystems_), - availability(availability_), last_osd_epoch(last_osd_epoch_), last_gwmap_epoch(last_gwmap_epoch_) + availability(availability_), last_osd_epoch(last_osd_epoch_), + last_gwmap_epoch(last_gwmap_epoch_), sequence(sequence_) { set_priority(CEPH_MSG_PRIO_HIGH); } @@ -77,6 +85,7 @@ public: const epoch_t& get_last_osd_epoch() const { return last_osd_epoch; } const epoch_t& get_last_gwmap_epoch() const { return last_gwmap_epoch; } const BeaconSubsystems& get_subsystems() const { return subsystems; }; + uint64_t get_sequence() const { return sequence; } private: ~MNVMeofGwBeacon() final {} @@ -91,10 +100,14 @@ public: encode(gw_id, payload); encode(gw_pool, payload); encode(gw_group, payload); - encode(subsystems, payload); + encode(subsystems, payload, features); encode((uint32_t)availability, payload); encode(last_osd_epoch, payload); encode(last_gwmap_epoch, payload); + // Only encode sequence for enhanced beacons (HEAD_VERSION >= 2) + if (get_header().version >= BEACON_VERSION_ENHANCED) { + encode(sequence, payload); + } } void decode_payload() override { @@ -111,6 +124,12 @@ public: availability = static_cast(tmp); decode(last_osd_epoch, p); decode(last_gwmap_epoch, p); + // Only decode sequence for enhanced beacons (HEAD_VERSION >= 2) + if (get_header().version >= BEACON_VERSION_ENHANCED && !p.end()) { + decode(sequence, p); + } else { + sequence = 0; // Legacy beacons don't have sequence field + } } private: diff --git a/src/mon/NVMeofGwSerialize.h b/src/mon/NVMeofGwSerialize.h index 9d43f242ae8..21e1dcc052a 100755 --- a/src/mon/NVMeofGwSerialize.h +++ b/src/mon/NVMeofGwSerialize.h @@ -19,6 +19,11 @@ #define dout_prefix *_dout << MODULE_PREFFIX << __PRETTY_FUNCTION__ << " " #define MAX_SUPPORTED_ANA_GROUPS 16 +// Local version constants that match MNVMeofGwBeacon.h +// We can't include that header here due to circular dependencies +#define BEACON_SUBSYS_VERSION_LEGACY 1 // Legacy beacon format (no diff support) +#define BEACON_SUBSYS_VERSION_ENHANCED 2 // Enhanced beacon format (with diff support) + inline std::ostream& operator<<( std::ostream& os, const gw_exported_states_per_group_t value) { switch (value) { @@ -106,7 +111,8 @@ inline std::ostream& operator<<(std::ostream& os, const BeaconListener value) { } inline std::ostream& operator<<(std::ostream& os, const BeaconSubsystem value) { - os << "BeaconSubsystem( nqn:" << value.nqn << ", listeners [ "; + os << "BeaconSubsystem( nqn:" << value.nqn << " descr " + << (uint32_t)value.change_descriptor << ", listeners [ "; for (const auto& list: value.listeners) os << list << " "; os << "] namespaces [ "; for (const auto& ns: value.namespaces) os << ns << " "; @@ -123,7 +129,9 @@ inline std::ostream& operator<<( std::ostream& os, const NvmeGwClientState value) { os << "NvmeGwState { group id: " << value.group_id << " gw_map_epoch " << value.gw_map_epoch - << " availablilty "<< value.availability + << " availablilty " << value.availability + << " sequence " << value.last_beacon_seq_number + << " sequence-ooo " << value.last_beacon_seq_ooo << " GwSubsystems: [ "; for (const auto& sub: value.subsystems) { os << sub.second << " "; @@ -273,7 +281,7 @@ inline void encode( for (const auto& sub: subsystems) { encode(sub.second.nqn, bl); if (version == 1) { - dout(20) << "encode ana_state vector version1 = " << version << dendl; + dout(20) << "encode ana_state vector version1 = " << (int)version << dendl; /* Version 1 requires exactly 16 entries */ ana_state_t filled(sub.second.ana_state); filled.resize( @@ -283,7 +291,7 @@ inline void encode( 0)); encode(filled, bl); } else { - dout(20) << "encode ana_state vector version2 = " << version << dendl; + dout(20) << "encode ana_state vector version2 = " << (int)version << dendl; encode(sub.second.ana_state, bl); } } @@ -307,23 +315,35 @@ inline void decode( } inline void encode(const NvmeGwClientState& state, ceph::bufferlist &bl, uint64_t features) { - ENCODE_START(1, 1, bl); + uint8_t version = 1; + + ENCODE_START(version, version, bl); encode(state.group_id, bl); encode(state.gw_map_epoch, bl); encode (state.subsystems, bl, features); encode((uint32_t)state.availability, bl); + if (version >= 2) { + encode((uint64_t)state.last_beacon_seq_number, bl); + encode(state.last_beacon_seq_ooo, bl); + } ENCODE_FINISH(bl); } inline void decode( NvmeGwClientState& state, ceph::bufferlist::const_iterator& bl) { - DECODE_START(1, bl); + DECODE_START(2, bl); decode(state.group_id, bl); decode(state.gw_map_epoch, bl); decode(state.subsystems, bl); uint32_t avail; + uint64_t last_beacon_seq_number; decode(avail, bl); state.availability = (gw_availability_t)avail; + if (struct_v >= 2) { + decode(last_beacon_seq_number, bl); + state.last_beacon_seq_number = last_beacon_seq_number; + decode(state.last_beacon_seq_ooo, bl); + } DECODE_FINISH(bl); } @@ -417,6 +437,7 @@ inline void encode(const NvmeAnaNonceMap& nonce_map, ceph::bufferlist &bl, uint64_t features) { ENCODE_START(1, 1, bl); encode((uint32_t)nonce_map.size(), bl); + dout(20) << "encode nonce map size " << nonce_map.size() << dendl; for (auto& ana_group_nonces : nonce_map) { // ana group id encode(ana_group_nonces.first, bl); @@ -457,10 +478,12 @@ inline void encode(const NvmeGwMonStates& gws, ceph::bufferlist &bl, version = 3; } ENCODE_START(version, version, bl); + dout(20) << "encode NvmeGwMonStates. struct_v: " << (int)version << dendl; encode ((uint32_t)gws.size(), bl); // number of gws in the group for (auto& gw : gws) { encode(gw.first, bl);// GW_id encode(gw.second.ana_grp_id, bl); // GW owns this group-id + dout(20) << "encode gw-id " << gw.first << dendl; if (version >= 2) { encode((uint32_t)gw.second.sm_state.size(), bl); for (auto &state_it:gw.second.sm_state) { @@ -470,7 +493,9 @@ inline void encode(const NvmeGwMonStates& gws, ceph::bufferlist &bl, encode((uint32_t)gw.second.availability, bl); encode((uint16_t)gw.second.performed_full_startup, bl); encode((uint16_t)gw.second.last_gw_map_epoch_valid, bl); - encode(gw.second.subsystems, bl); + dout(20) << "encode availability " << gw.second.availability + << " startup " << (int)gw.second.performed_full_startup << dendl; + encode(gw.second.subsystems, bl, features); encode((uint32_t)gw.second.blocklist_data.size(), bl); for (auto &blklst_itr: gw.second.blocklist_data) { @@ -487,7 +512,7 @@ inline void encode(const NvmeGwMonStates& gws, ceph::bufferlist &bl, encode((uint32_t)gw.second.availability, bl); encode((uint16_t)gw.second.performed_full_startup, bl); encode((uint16_t)gw.second.last_gw_map_epoch_valid, bl); - encode(gw.second.subsystems, bl); // TODO reuse but put features - encode version + encode(gw.second.subsystems, bl, features); Blocklist_data bl_data[MAX_SUPPORTED_ANA_GROUPS]; for (auto &blklst_itr: gw.second.blocklist_data) { bl_data[blklst_itr.first].osd_epoch = blklst_itr.second.osd_epoch; @@ -589,6 +614,7 @@ inline void decode( dout(20) << "decode addr_vect and beacon_index" << dendl; gw_created.addr_vect.decode(bl); decode(gw_created.beacon_index, bl); + dout(20) << "decoded beacon_index " << gw_created.beacon_index << dendl; } gws[gw_name] = gw_created; @@ -794,7 +820,7 @@ inline void decode(NvmeGwTimers& md, ceph::buffer::list::const_iterator &bl) { } inline void encode(const BeaconNamespace& ns, ceph::bufferlist &bl) { - ENCODE_START(1, 1, bl); + ENCODE_START(BEACON_SUBSYS_VERSION_LEGACY, BEACON_SUBSYS_VERSION_LEGACY, bl); encode(ns.anagrpid, bl); encode(ns.nonce, bl); ENCODE_FINISH(bl); @@ -808,7 +834,7 @@ inline void decode(BeaconNamespace& ns, ceph::buffer::list::const_iterator &bl) } inline void encode(const BeaconListener& ls, ceph::bufferlist &bl) { - ENCODE_START(1, 1, bl); + ENCODE_START(BEACON_SUBSYS_VERSION_LEGACY, BEACON_SUBSYS_VERSION_LEGACY, bl); encode(ls.address_family, bl); encode(ls.address, bl); encode(ls.svcid, bl); @@ -823,22 +849,39 @@ inline void decode(BeaconListener& ls, ceph::buffer::list::const_iterator &bl) { DECODE_FINISH(bl); } -inline void encode(const BeaconSubsystem& sub, ceph::bufferlist &bl) { - ENCODE_START(1, 1, bl); +inline void encode(const BeaconSubsystem& sub, ceph::bufferlist &bl, uint64_t features) { + uint8_t version = BEACON_SUBSYS_VERSION_LEGACY; // Default to legacy version + + // For legacy encoding, skip deleted subsystems to maintain compatibility + if (version == BEACON_SUBSYS_VERSION_LEGACY && + sub.change_descriptor != subsystem_change_t::SUBSYSTEM_ADDED) { + dout(4) << "encode BeaconSubsystem: skipping subsystem " << sub.nqn + << " with change_descriptor " << (int)sub.change_descriptor + << " in legacy mode" << dendl; + return; // Skip encoding this subsystem entirely + } + + ENCODE_START(version, version, bl); encode(sub.nqn, bl); + dout(20) << "encode BeaconSubsystems " << sub.nqn <<" features " << features + << " version " << (int)version << dendl; encode((uint32_t)sub.listeners.size(), bl); for (const auto& ls: sub.listeners) encode(ls, bl); encode((uint32_t)sub.namespaces.size(), bl); for (const auto& ns: sub.namespaces) encode(ns, bl); + if (version >= BEACON_SUBSYS_VERSION_ENHANCED) { + encode((uint32_t)sub.change_descriptor, bl); + dout(20) << "encode BeaconSubsystems change-descr: " << (uint32_t)sub.change_descriptor << dendl; + } ENCODE_FINISH(bl); } inline void decode(BeaconSubsystem& sub, ceph::buffer::list::const_iterator &bl) { - DECODE_START(1, bl); - dout(20) << "decode BeaconSubsystems " << dendl; + DECODE_START(BEACON_SUBSYS_VERSION_ENHANCED, bl); // Always decode with enhanced version support decode(sub.nqn, bl); + dout(20) << "decode BeaconSubsystems " << sub.nqn << dendl; uint32_t s; sub.listeners.clear(); decode(s, bl); @@ -855,6 +898,12 @@ inline void decode(BeaconSubsystem& sub, ceph::buffer::list::const_iterator &bl) decode(ns, bl); sub.namespaces.push_back(ns); } + if (struct_v >= BEACON_SUBSYS_VERSION_ENHANCED) { + uint32_t change_desc; + decode(change_desc, bl); + sub.change_descriptor = static_cast(change_desc); + dout(20) << "decode BeaconSubsystems version >= " << BEACON_SUBSYS_VERSION_ENHANCED << dendl; + } DECODE_FINISH(bl); } diff --git a/src/mon/NVMeofGwTypes.h b/src/mon/NVMeofGwTypes.h index 097397795a9..8ee27d49efd 100755 --- a/src/mon/NVMeofGwTypes.h +++ b/src/mon/NVMeofGwTypes.h @@ -17,6 +17,12 @@ #include #include #include +#include +#include +#include +#include +#include "include/types.h" +#include "msg/msg_types.h" using NvmeGwId = std::string; using NvmeGroupKey = std::pair; @@ -46,6 +52,12 @@ enum class gw_availability_t { GW_DELETED }; +enum class subsystem_change_t { + SUBSYSTEM_ADDED, + SUBSYSTEM_CHANGED, + SUBSYSTEM_DELETED +}; + #define REDUNDANT_GW_ANA_GROUP_ID 0xFF using SmState = std::map < NvmeAnaGrpId, gw_states_per_group_t>; @@ -85,6 +97,7 @@ struct BeaconSubsystem { NvmeNqnId nqn; std::list listeners; std::list namespaces; + subsystem_change_t change_descriptor = subsystem_change_t::SUBSYSTEM_ADDED; // Define the equality operator bool operator==(const BeaconSubsystem& other) const { @@ -130,7 +143,9 @@ struct NvmeGwMonState { BlocklistData blocklist_data; //ceph entity address allocated for the GW-client that represents this GW-id entity_addrvec_t addr_vect; - uint16_t beacon_index = 0; + uint64_t beacon_sequence = 0;// sequence number of last beacon copied to GW state + bool beacon_sequence_ooo = false; // last beacon sequence was out of order; + uint16_t beacon_index = 0; // used for filter acks sent to the client as response to beacon /** * during redeploy action and maybe other emergency use-cases gw performs scenario * that we call fast-reboot. It quickly reboots(due to redeploy f.e) and sends the @@ -169,6 +184,9 @@ struct NvmeGwMonState { // it expects it performed the full startup performed_full_startup = false; } + void reset_beacon_sequence(){ + beacon_sequence = 0; + } void standby_state(NvmeAnaGrpId grpid) { sm_state[grpid] = gw_states_per_group_t::GW_STANDBY_STATE; } @@ -230,6 +248,8 @@ struct NvmeGwClientState { epoch_t gw_map_epoch; GwSubsystems subsystems; gw_availability_t availability; + uint64_t last_beacon_seq_number; + bool last_beacon_seq_ooo; //out of order sequence NvmeGwClientState(NvmeAnaGrpId id, epoch_t epoch, gw_availability_t available) : group_id(id), gw_map_epoch(epoch), availability(available) {} diff --git a/src/msg/Message.cc b/src/msg/Message.cc index db5b1f573ff..1395b0ba507 100644 --- a/src/msg/Message.cc +++ b/src/msg/Message.cc @@ -897,7 +897,7 @@ Message *decode_message(CephContext *cct, case MSG_MNVMEOF_GW_BEACON: m = make_message(); - break; + break; case MSG_MON_MGR_REPORT: m = make_message(); diff --git a/src/nvmeof/NVMeofGwMonitorClient.cc b/src/nvmeof/NVMeofGwMonitorClient.cc index 2000f718a7d..acac5abda33 100644 --- a/src/nvmeof/NVMeofGwMonitorClient.cc +++ b/src/nvmeof/NVMeofGwMonitorClient.cc @@ -19,6 +19,7 @@ #include "include/compat.h" #include "include/stringify.h" +#include "include/ceph_features.h" #include "global/global_context.h" #include "global/signal_handler.h" @@ -28,6 +29,7 @@ #include "NVMeofGwMonitorClient.h" #include "NVMeofGwClient.h" #include "NVMeofGwMonitorGroupClient.h" +#include "nvmeof/NVMeofGwUtils.h" #define dout_context g_ceph_context #define dout_subsys ceph_subsys_mon @@ -41,6 +43,8 @@ NVMeofGwMonitorClient::NVMeofGwMonitorClient(int argc, const char **argv) : last_map_time(std::chrono::steady_clock::now()), reset_timestamp(std::chrono::steady_clock::now()), start_time(last_map_time), + cluster_features(0), + poolctx(), monc{g_ceph_context, poolctx}, client_messenger(Messenger::create(g_ceph_context, "async", entity_name_t::CLIENT(-1), "client", getpid())), objecter{g_ceph_context, client_messenger.get(), &monc, poolctx}, @@ -211,7 +215,7 @@ void NVMeofGwMonitorClient::send_beacon() { ceph_assert(ceph_mutex_is_locked_by_me(beacon_lock)); gw_availability_t gw_availability = gw_availability_t::GW_CREATED; - BeaconSubsystems subs; + BeaconSubsystems current_subsystems; NVMeofGwClient gw_client( grpc::CreateChannel(gateway_address, gw_creds())); subsystems_info gw_subsystems; @@ -231,26 +235,42 @@ void NVMeofGwMonitorClient::send_beacon() BeaconListener bls = { ls.adrfam(), ls.traddr(), ls.trsvcid() }; bsub.listeners.push_back(bls); } - subs.push_back(bsub); + current_subsystems.push_back(bsub); } } + // Determine change descriptors by comparing with previous beacon's subsystems + BeaconSubsystems subs = current_subsystems; + determine_subsystem_changes(prev_beacon_subsystems, subs); + auto group_key = std::make_pair(pool, group); NvmeGwClientState old_gw_state; // if already got gateway state in the map if (first_beacon == false && get_gw_state("old map", map, group_key, name, old_gw_state)) gw_availability = ok ? gw_availability_t::GW_AVAILABLE : gw_availability_t::GW_UNAVAILABLE; - dout(10) << "sending beacon as gid " << monc.get_global_id() << " availability " << (int)gw_availability << + dout(1) << "sending beacon as gid " << monc.get_global_id() << " availability " << (int)gw_availability << " osdmap_epoch " << osdmap_epoch << " gwmap_epoch " << gwmap_epoch << dendl; + + // Check if NVMEOF_BEACON_DIFF feature is supported by the cluster + bool include_diff = true; + + // Send beacon with appropriate version based on cluster features auto m = ceph::make_message( name, pool, group, - subs, + include_diff ? subs : current_subsystems, gw_availability, osdmap_epoch, - gwmap_epoch); + gwmap_epoch, + beacon_sequence, + include_diff // Pass the feature flag directly to constructor + ); + dout(10) << "sending beacon with diff support: " << (include_diff ? "enabled" : "disabled") << dendl; + monc.send_mon_message(std::move(m)); + ++beacon_sequence; + prev_beacon_subsystems = std::move(current_subsystems); } void NVMeofGwMonitorClient::disconnect_panic() @@ -352,6 +372,22 @@ void NVMeofGwMonitorClient::handle_nvmeof_gw_map(ceph::ref_t nmap) // ensure that the gateway state has not vanished ceph_assert(got_new_gw_state || !got_old_gw_state); + // Check if the last_beacon_seq_number in the received map doesn't match our last sent beacon_sequence + // beacon_sequence is incremented after sending, so we compare with (beacon_sequence - 1) + { + std::lock_guard bl(beacon_lock); + if (got_new_gw_state && new_gw_state.last_beacon_seq_ooo) { + //new_gw_state.last_beacon_seq_number != (beacon_sequence - 1)) { + dout(4) << "Beacon sequence mismatch detected. Expected: " << (beacon_sequence - 1) + << ", received: " << new_gw_state.last_beacon_seq_number + << ". Truncating previous subsystems list." << dendl; + beacon_sequence = new_gw_state.last_beacon_seq_number + 1; + prev_beacon_subsystems.clear(); + dout(4) << "OOO map received, Ignore it" << dendl; + return; + } + } + if (!got_old_gw_state) { if (!got_new_gw_state) { dout(10) << "Can not find new gw state" << dendl; @@ -386,10 +422,29 @@ void NVMeofGwMonitorClient::handle_nvmeof_gw_map(ceph::ref_t nmap) } } + // Combined subsystems + const auto initial_ana_state = std::make_pair(gw_exported_states_per_group_t::GW_EXPORTED_INACCESSIBLE_STATE, (epoch_t)0); + GwSubsystems combined_subsystems = new_gw_state.subsystems; + for (const auto& nqn_state_pair: old_gw_state.subsystems) { + const auto& nqn = nqn_state_pair.first; + auto& old_nqn_state = nqn_state_pair.second; + + // The monitor might remove active subsystems from the new distributed GwSubsystems. + // In such cases, ensure an INACCESSIBLE state is generated for subsystems + // that were present in the old state but are now missing. + if (new_gw_state.subsystems.find(nqn) == new_gw_state.subsystems.end()) { + ana_state_t all_disabled(old_nqn_state.ana_state.size(), initial_ana_state); + dout(4) << "set all groups to Inacccessible stat for " << nqn << dendl; + NqnState nqn_state(nqn, all_disabled); + + combined_subsystems.insert({nqn, nqn_state}); + } + } + // Gather all state changes ana_info ai; epoch_t max_blocklist_epoch = 0; - for (const auto& nqn_state_pair: new_gw_state.subsystems) { + for (const auto& nqn_state_pair: combined_subsystems) { auto& sub = nqn_state_pair.second; const auto& nqn = nqn_state_pair.first; nqn_ana_states nas; @@ -403,7 +458,6 @@ void NVMeofGwMonitorClient::handle_nvmeof_gw_map(ceph::ref_t nmap) sub.ana_state.size(); for (NvmeAnaGrpId ana_grp_index = 0; ana_grp_index < ana_state_size; ana_grp_index++) { - const auto initial_ana_state = std::make_pair(gw_exported_states_per_group_t::GW_EXPORTED_INACCESSIBLE_STATE, (epoch_t)0); auto new_group_state = (ana_grp_index < sub.ana_state.size()) ? sub.ana_state[ana_grp_index] : initial_ana_state; @@ -457,6 +511,17 @@ Dispatcher::dispatch_result_t NVMeofGwMonitorClient::ms_dispatch2(const ref_tget_type() << dendl; + // print connection features for all incoming messages and update cluster features + if (m->get_connection()) { + uint64_t features = m->get_connection()->get_features(); + dout(4) << "Monitor connection features: 0x" << std::hex << features << std::dec << dendl; + + // Update cluster features with the union of all seen features + // This ensures we track the highest level of features supported by the cluster + cluster_features |= features; + dout(10) << "Updated cluster features: 0x" << std::hex << cluster_features << std::dec << dendl; + } + if (m->get_type() == MSG_MNVMEOF_GW_MAP) { handle_nvmeof_gw_map(ref_cast(m)); return Dispatcher::HANDLED(); diff --git a/src/nvmeof/NVMeofGwMonitorClient.h b/src/nvmeof/NVMeofGwMonitorClient.h index f187a47b481..c31ca6f1113 100644 --- a/src/nvmeof/NVMeofGwMonitorClient.h +++ b/src/nvmeof/NVMeofGwMonitorClient.h @@ -52,7 +52,9 @@ private: bool first_beacon = true; bool set_group_id = false; - + uint64_t beacon_sequence = 0; + BeaconSubsystems prev_beacon_subsystems; + uint64_t cluster_features = 0; // track cluster features for beacon encoding // init gw ssl opts void init_gw_ssl_opts(); @@ -75,7 +77,7 @@ protected: void send_config_beacon(); void send_beacon(); - + public: NVMeofGwMonitorClient(int argc, const char **argv); ~NVMeofGwMonitorClient() override; diff --git a/src/nvmeof/NVMeofGwUtils.cc b/src/nvmeof/NVMeofGwUtils.cc new file mode 100644 index 00000000000..9beccc8668c --- /dev/null +++ b/src/nvmeof/NVMeofGwUtils.cc @@ -0,0 +1,53 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2025 IBM, Inc. + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + */ + +#include "nvmeof/NVMeofGwUtils.h" + +void determine_subsystem_changes(const BeaconSubsystems& old_subsystems, + BeaconSubsystems& new_subsystems) { + BeaconSubsystems result; + + // for each subsystem in new_subsystems, check if it's added or changed + for (const auto& new_sub : new_subsystems) { + auto old_it = std::find_if(old_subsystems.begin(), old_subsystems.end(), + [&](const BeaconSubsystem& s) { return s.nqn == new_sub.nqn; }); + if (old_it == old_subsystems.end()) { + // Subsystem not found in old list - it's new + BeaconSubsystem added = new_sub; + added.change_descriptor = subsystem_change_t::SUBSYSTEM_ADDED; + result.push_back(std::move(added)); + } else { + // subsystem exists - check if it changed + if (!(*old_it == new_sub)) { + BeaconSubsystem changed = new_sub; + changed.change_descriptor = subsystem_change_t::SUBSYSTEM_CHANGED; + result.push_back(std::move(changed)); + } + // else: unchanged, do not add + } + } + + // for any subsystem in old_subsystems not present in new_subsystems, add as deleted + for (const auto& old_sub : old_subsystems) { + auto found = std::find_if(new_subsystems.begin(), new_subsystems.end(), + [&](const BeaconSubsystem& s) { return s.nqn == old_sub.nqn; }); + if (found == new_subsystems.end()) { + BeaconSubsystem deleted_sub = old_sub; + deleted_sub.change_descriptor = subsystem_change_t::SUBSYSTEM_DELETED; + result.push_back(std::move(deleted_sub)); + } + } + + new_subsystems = std::move(result); +} + diff --git a/src/nvmeof/NVMeofGwUtils.h b/src/nvmeof/NVMeofGwUtils.h new file mode 100644 index 00000000000..a9c25317d81 --- /dev/null +++ b/src/nvmeof/NVMeofGwUtils.h @@ -0,0 +1,23 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2025 IBM, Inc. + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + */ + +#ifndef __NVMEOFGWUTILS_H__ +#define __NVMEOFGWUTILS_H__ +#include "mon/NVMeofGwTypes.h" +#include + +// utility for diffing nvmeof subsystems changes +void determine_subsystem_changes(const BeaconSubsystems& old_subsystems, + BeaconSubsystems& new_subsystems); + +#endif diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index a22c7ccb236..9e190653b7f 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -1043,3 +1043,11 @@ add_ceph_unittest(unittest_ceph_assert) target_link_libraries(unittest_ceph_assert ceph-common global) endif() +add_executable(test_nvmeof_gw_utils + test_nvmeof_gw_utils.cc + ../nvmeof/NVMeofGwUtils.cc + ) +target_link_libraries(test_nvmeof_gw_utils + mon ceph-common global-static + ) + diff --git a/src/test/test_nvmeof_gw_utils.cc b/src/test/test_nvmeof_gw_utils.cc new file mode 100644 index 00000000000..c7d2ccfa561 --- /dev/null +++ b/src/test/test_nvmeof_gw_utils.cc @@ -0,0 +1,69 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2025 IBM, Inc. + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ + +#include "nvmeof/NVMeofGwUtils.h" +#include "mon/NVMeofGwTypes.h" +#include +#include +#include "include/ceph_assert.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_mon +#undef dout_prefix +#define dout_prefix *_dout + +void test_determine_subsystem_changes() { + std::cout << __func__ << "\n\n" << std::endl; + // Prepare old and new subsystems + BeaconSubsystem sub1_old = { "nqn1", {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; + BeaconSubsystem sub2_old = { "nqn2", {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; + BeaconSubsystem sub3_old = { "nqn3", {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; + BeaconSubsystems old_subs = { sub1_old, sub2_old, sub3_old }; + + // sub1 unchanged, sub2 changed, sub4 added, sub3 deleted + BeaconSubsystem sub1_new = { "nqn1", {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; + BeaconSubsystem sub2_new = { "nqn2", { {"IPv4", "1.2.3.4", "4420"} }, {}, subsystem_change_t::SUBSYSTEM_ADDED }; // changed listeners + BeaconSubsystem sub4_new = { "nqn4", {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; + BeaconSubsystems new_subs = { sub1_new, sub2_new, sub4_new }; + + determine_subsystem_changes(old_subs, new_subs); + + // After call, new_subs should only contain changed, added, and deleted subsystems + // sub1 (unchanged) should be removed + // sub2 (changed) should be present with SUBSYSTEM_CHANGED + // sub4 (added) should be present with SUBSYSTEM_ADDED + // sub3 (deleted) should be present with SUBSYSTEM_DELETED + bool found_sub2 = false, found_sub3 = false, found_sub4 = false; + for (const auto& s : new_subs) { + if (s.nqn == "nqn2") { + found_sub2 = true; + ceph_assert(s.change_descriptor == subsystem_change_t::SUBSYSTEM_CHANGED); + } else if (s.nqn == "nqn3") { + found_sub3 = true; + ceph_assert(s.change_descriptor == subsystem_change_t::SUBSYSTEM_DELETED); + } else if (s.nqn == "nqn4") { + found_sub4 = true; + ceph_assert(s.change_descriptor == subsystem_change_t::SUBSYSTEM_ADDED); + } else { + ceph_assert(false && "Unexpected subsystem in result"); + } + } + ceph_assert(found_sub2 && found_sub3 && found_sub4); + std::cout << "determine_subsystem_changes test passed" << std::endl; +} + +int main(int argc, const char **argv) { + test_determine_subsystem_changes(); + return 0; +} diff --git a/src/test/test_nvmeof_mon_encoding.cc b/src/test/test_nvmeof_mon_encoding.cc index 704342c4c16..d3f1c47c009 100644 --- a/src/test/test_nvmeof_mon_encoding.cc +++ b/src/test/test_nvmeof_mon_encoding.cc @@ -35,12 +35,17 @@ void test_NVMeofGwMap() { std::string pool = "pool1"; std::string group = "grp1"; auto group_key = std::make_pair(pool, group); - pending_map.cfg_add_gw("GW1" ,group_key); - pending_map.cfg_add_gw("GW2" ,group_key); - pending_map.cfg_add_gw("GW3" ,group_key); + std::string nqn = "nqn-nqn"; + BeaconSubsystem sub = { nqn, {}, {}, subsystem_change_t::SUBSYSTEM_CHANGED }; + BeaconSubsystems subs = {sub}; + + pending_map.cfg_add_gw("GW1" ,group_key, true); + pending_map.cfg_add_gw("GW2" ,group_key, true); + pending_map.cfg_add_gw("GW3" ,group_key, true); NvmeNonceVector new_nonces = {"abc", "def","hij"}; pending_map.created_gws[group_key]["GW1"].nonce_map[1] = new_nonces; pending_map.created_gws[group_key]["GW1"].performed_full_startup = true; + pending_map.created_gws[group_key]["GW1"].subsystems = subs; int i = 0; for (auto & blklst_itr : pending_map.created_gws[group_key]["GW1"].blocklist_data){ blklst_itr.second.osd_epoch = 2*(i++); @@ -52,21 +57,25 @@ void test_NVMeofGwMap() { dout(0) << pending_map << dendl; ceph::buffer::list bl; + dout(0) << pending_map.created_gws[group_key]["GW1"].subsystems << dendl; + pending_map.encode(bl, CEPH_FEATURES_ALL); auto p = bl.cbegin(); pending_map.decode(p); dout(0) << " == Dump map after Decode: == " < map; std::string pool = "pool1"; std::string group = "grp1"; std::string gw_id = "GW1"; - NvmeGwClientState state(1, 32, gw_availability_t::GW_UNAVAILABLE); + NvmeGwClientState state(1, 32, gw_availability_t::GW_UNAVAILABLE, 0, false); std::string nqn = "nqn"; ana_state_t ana_state; NqnState nqn_state(nqn, ana_state); @@ -81,9 +90,9 @@ void test_MNVMeofGwMap() { encode(map, bl, CEPH_FEATURES_ALL); dout(0) << "encoded: " << map << dendl; decode(map, bl); - dout(0) << "decode: " << map << dendl; + dout(0) << "decoded: " << map << dendl; - BeaconSubsystem sub = { nqn, {}, {} }; + BeaconSubsystem sub = { nqn, {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; NVMeofGwMap pending_map; pending_map.epoch = 2; auto msg1 = make_message(pending_map); @@ -94,12 +103,13 @@ void test_MNVMeofGwMap() { int epoch = msg1->get_gwmap_epoch(); dout(0) << "after decode empty msg: " << *msg1 << " epoch " << epoch << dendl; - pending_map.cfg_add_gw("GW1" ,group_key); - pending_map.cfg_add_gw("GW2" ,group_key); - pending_map.cfg_add_gw("GW3" ,group_key); + pending_map.cfg_add_gw("GW1" ,group_key, true); + pending_map.cfg_add_gw("GW2" ,group_key, true); + pending_map.cfg_add_gw("GW3" ,group_key, true); NvmeNonceVector new_nonces = {"abc", "def","hij"}; pending_map.created_gws[group_key]["GW1"].nonce_map[1] = new_nonces; pending_map.created_gws[group_key]["GW1"].subsystems.push_back(sub); + int i = 0; for (auto & blklst_itr : pending_map.created_gws[group_key]["GW1"].blocklist_data){ blklst_itr.second.osd_epoch = 2*(i++); @@ -115,7 +125,7 @@ void test_MNVMeofGwMap() { dout(0) << "after encode msg: " << *msg << dendl; msg->decode_payload(); dout(0) << "after decode msg: " << *msg << dendl; - + //dout(0) << "\n == Test GW Delete ==" << dendl; //pending_map.cfg_delete_gw("GW1" ,group_key); //dout(0) << "deleted GW1 " << pending_map << dendl; @@ -127,10 +137,10 @@ void test_MNVMeofGwMap() { //dout(0) << "deleted GW2 " << pending_map << dendl; //dout(0) << "delete of wrong gw id" << dendl; - //pending_map.cfg_delete_gw("wow" ,group_key); + //pending_map.cfg_delete_gw("wow" ,group_key, true); - pending_map.cfg_delete_gw("GW3" ,group_key); - dout(0) << "deleted GW3 . we should see the empty map " << pending_map << dendl; + //pending_map.cfg_delete_gw("GW3" ,group_key); + //dout(0) << "deleted GW3 . we should see the empty map " << pending_map << dendl; } @@ -141,11 +151,13 @@ void test_MNVMeofGwBeacon() { std::string gw_group = "group"; gw_availability_t availability = gw_availability_t::GW_AVAILABLE; std::string nqn = "nqn"; - BeaconSubsystem sub = { nqn, {}, {} }; + BeaconSubsystem sub = { nqn, {}, {}, subsystem_change_t::SUBSYSTEM_CHANGED }; BeaconSubsystems subs = { sub }; epoch_t osd_epoch = 17; epoch_t gwmap_epoch = 42; - + uint64_t sequence = 12345; + + // Test legacy beacon (without diff support) auto msg = make_message( gw_id, gw_pool, @@ -153,22 +165,87 @@ void test_MNVMeofGwBeacon() { subs, availability, osd_epoch, - gwmap_epoch); - msg->encode_payload(0); + gwmap_epoch + // sequence defaults to 0 + // enable_diff defaults to false + ); + msg->encode_payload(CEPH_FEATURES_ALL); msg->decode_payload(); - dout(0) << "decode msg: " << *msg << dendl; + dout(0) << "decode msg (revision 1): " << *msg << dendl; ceph_assert(msg->get_gw_id() == gw_id); ceph_assert(msg->get_gw_pool() == gw_pool); ceph_assert(msg->get_gw_group() == gw_group); ceph_assert(msg->get_availability() == availability); ceph_assert(msg->get_last_osd_epoch() == osd_epoch); ceph_assert(msg->get_last_gwmap_epoch() == gwmap_epoch); + // Legacy beacons don't preserve sequence field - it gets reset to 0 + ceph_assert(msg->get_sequence() == 0); const auto& dsubs = msg->get_subsystems(); auto it = std::find_if(dsubs.begin(), dsubs.end(), [&nqn](const auto& element) { return element.nqn == nqn; }); ceph_assert(it != dsubs.end()); + ceph_assert(it->change_descriptor == subsystem_change_t::SUBSYSTEM_CHANGED); + + // Test enhanced beacon (with diff support) + auto msg2 = make_message( + gw_id, + gw_pool, + gw_group, + subs, + availability, + osd_epoch, + gwmap_epoch, + sequence, + true // enable_diff = true + ); + msg2->encode_payload(CEPH_FEATURES_ALL); + msg2->decode_payload(); + dout(0) << "decode msg (revision 2): " << *msg2 << dendl; + ceph_assert(msg2->get_gw_id() == gw_id); + ceph_assert(msg2->get_gw_pool() == gw_pool); + ceph_assert(msg2->get_gw_group() == gw_group); + ceph_assert(msg2->get_availability() == availability); + ceph_assert(msg2->get_last_osd_epoch() == osd_epoch); + ceph_assert(msg2->get_last_gwmap_epoch() == gwmap_epoch); + ceph_assert(msg2->get_sequence() == sequence); + const auto& dsubs2 = msg2->get_subsystems(); + auto it2 = std::find_if(dsubs2.begin(), dsubs2.end(), + [&nqn](const auto& element) { + return element.nqn == nqn; + }); + ceph_assert(it2 != dsubs2.end()); + ceph_assert(it2->change_descriptor == subsystem_change_t::SUBSYSTEM_CHANGED); +} + +void test_subsystem_change_descriptors() { + dout(0) << __func__ << "\n\n" << dendl; + // Test different change descriptors + BeaconSubsystem sub1 = { "nqn1", {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; + BeaconSubsystem sub2 = { "nqn2", {}, {}, subsystem_change_t::SUBSYSTEM_CHANGED }; + BeaconSubsystem sub3 = { "nqn3", {}, {}, subsystem_change_t::SUBSYSTEM_ADDED }; + BeaconSubsystems subs = { sub1, sub2, sub3 }; + // Encode and decode + ceph::buffer::list bl; + encode(subs, bl, CEPH_FEATURES_ALL); + auto p = bl.cbegin(); + BeaconSubsystems decoded_subs; + decode(decoded_subs, p); + // Verify change descriptors are preserved + auto it1 = std::find_if(decoded_subs.begin(), decoded_subs.end(), + [](const auto& s) { return s.nqn == "nqn1"; }); + auto it2 = std::find_if(decoded_subs.begin(), decoded_subs.end(), + [](const auto& s) { return s.nqn == "nqn2"; }); + auto it3 = std::find_if(decoded_subs.begin(), decoded_subs.end(), + [](const auto& s) { return s.nqn == "nqn3"; }); + ceph_assert(it1 != decoded_subs.end()); + ceph_assert(it2 != decoded_subs.end()); + ceph_assert(it3 != decoded_subs.end()); + ceph_assert(it1->change_descriptor == subsystem_change_t::SUBSYSTEM_ADDED); + ceph_assert(it2->change_descriptor == subsystem_change_t::SUBSYSTEM_CHANGED); + ceph_assert(it3->change_descriptor == subsystem_change_t::SUBSYSTEM_ADDED); + dout(0) << "Subsystem change descriptors test passed" << dendl; } void test_NVMeofGwTimers() @@ -206,6 +283,7 @@ int main(int argc, const char **argv) test_NVMeofGwMap(); test_MNVMeofGwMap(); test_MNVMeofGwBeacon(); + test_subsystem_change_descriptors(); test_NVMeofGwTimers(); } -- 2.39.5