{expected_data});
}
-void ECPeeringTestFixture::run_parallel_recovery_and_verify_callbacks(
+// Helper function that performs the actual recovery logic
+// Must be called within event loop context on the primary OSD
+void ECPeeringTestFixture::do_run_parallel_recovery_and_verify_callbacks_impl(
const std::vector<std::string>& obj_names,
int target_osd,
- const std::vector<std::string>& expected_data)
+ const std::vector<std::string>& expected_data,
+ int primary_shard)
{
- // Verify we have matching sizes
- ASSERT_EQ(obj_names.size(), expected_data.size())
- << "obj_names and expected_data must have the same size";
-
- // Get the actual primary from the OSDMap
- int primary_shard = get_primary_shard_from_osdmap();
- if (primary_shard < 0 || primary_shard == CRUSH_ITEM_NONE) {
- // No valid primary, cannot run recovery
- return;
- }
auto primary_ps = get_peering_state(primary_shard);
pg_shard_t target_shard(target_osd, shard_id_t(target_osd));
<< "Object " << obj_names[i] << " should be in primary " << target_osd << "'s missing set";
std::cout << " OSD " << target_osd << " is the primary and has object " << obj_names[i] << " in its own missing set" << std::endl;
+ obcs.push_back(ObjectContextRef());
} else {
// The target OSD is a peer, check peer_missing
// Verify the missing item's need version matches what we read from the store
ASSERT_EQ(missing_item.need, oi.version)
<< "Missing item need version should match OI version from primary store for " << obj_names[i];
+
+ // Get OBC for this object - matches PrimaryLogPG::prep_object_replica_pushes behavior
+ // which calls get_object_context(soid, false) and handles null response
+ // Pass can_create=false to ensure we reload from disk with all attributes
+ ObjectContextRef obc = get_object_context(hoid, false);
+ ASSERT_TRUE(obc) << "Failed to load OBC from disk for " << obj_names[i];
+ ASSERT_FALSE(obc->attr_cache.empty())
+ << "OBC attr_cache must be populated for recovery of " << obj_names[i];
+ obcs.push_back(obc);
}
missing_items.push_back(missing_item);
-
- // Create OBC for this object
- ObjectContextRef obc = get_or_create_obc(hoid, true, expected_data[i].length());
- ASSERT_FALSE(obc->attr_cache.empty())
- << "OBC attr_cache must be populated for recovery of " << obj_names[i];
- obcs.push_back(obc);
}
// Reset recovery callback tracker before starting recovery
// Verify the recovered data
bufferlist read_bl;
- int r = read_object(obj_names[i], 0, expected_data[i].length(),
- read_bl, expected_data[i].length());
- EXPECT_EQ(r, (int)expected_data[i].length())
- << "Should read full object " << obj_names[i];
-
- std::string read_data(read_bl.c_str(), read_bl.length());
- EXPECT_EQ(read_data, expected_data[i])
- << "Recovered data should match for " << obj_names[i];
+ if (expected_data[i].size() > 0)
+ {
+ int r = read_object(obj_names[i], 0, expected_data[i].length(),
+ read_bl, expected_data[i].length());
+ EXPECT_EQ(r, (int)expected_data[i].length())
+ << "Should read full object " << obj_names[i];
+
+ std::string read_data(read_bl.c_str(), read_bl.length());
+ EXPECT_EQ(read_data, expected_data[i])
+ << "Recovered data should match for " << obj_names[i];
+ }
std::cout << " ✓ Object " << obj_names[i] << " recovered successfully" << std::endl;
}
std::cout << "\n === All parallel recovery callbacks and data verified successfully ===" << std::endl;
}
+
+// Public interface that schedules the recovery on the primary OSD
+void ECPeeringTestFixture::run_parallel_recovery_and_verify_callbacks(
+ const std::vector<std::string>& obj_names,
+ int target_osd,
+ const std::vector<std::string>& expected_data)
+{
+ // Verify we have matching sizes
+ ASSERT_EQ(obj_names.size(), expected_data.size())
+ << "obj_names and expected_data must have the same size";
+
+ // Get the actual primary from the OSDMap
+ int primary_shard = get_primary_shard_from_osdmap();
+ if (primary_shard < 0 || primary_shard == CRUSH_ITEM_NONE) {
+ // No valid primary, cannot run recovery
+ return;
+ }
+
+ // Schedule the recovery operation on the primary OSD
+ event_loop->schedule_transaction(primary_shard, [this, obj_names, target_osd, expected_data, primary_shard]() {
+ do_run_parallel_recovery_and_verify_callbacks_impl(obj_names, target_osd, expected_data, primary_shard);
+ });
+ event_loop->run_until_idle();
+}
const std::vector<std::string>& obj_names,
int target_osd,
const std::vector<std::string>& expected_data);
+
+private:
+ /**
+ * Implementation function for parallel recovery that runs within event loop context.
+ * This is called by the public wrapper function after scheduling on the primary OSD.
+ */
+ void do_run_parallel_recovery_and_verify_callbacks_impl(
+ const std::vector<std::string>& obj_names,
+ int target_osd,
+ const std::vector<std::string>& expected_data,
+ int instance);
};
ObjectContextRef get_obc(
const hobject_t &hoid,
const std::map<std::string, ceph::buffer::list, std::less<>> &attrs) override {
- return ObjectContextRef();
+ // This is called by the backend during recovery to create an OBC from attributes.
+ // Create a minimal OBC with the provided attributes.
+ ObjectContextRef obc = std::make_shared<ObjectContext>();
+
+ // Decode object_info from attributes
+ auto it_oi = attrs.find(OI_ATTR);
+ if (it_oi != attrs.end()) {
+ try {
+ bufferlist::const_iterator bliter = it_oi->second.begin();
+ decode(obc->obs.oi, bliter);
+ obc->obs.exists = true;
+ } catch (...) {
+ obc->obs.oi = object_info_t(hoid);
+ obc->obs.exists = false;
+ }
+ } else {
+ obc->obs.oi = object_info_t(hoid);
+ obc->obs.exists = false;
+ }
+
+ obc->ssc = nullptr;
+ obc->attr_cache = attrs;
+
+ return obc;
}
bool try_lock_for_read(
return completion_result;
}
-int PGBackendTestFixture::create_and_write(
+// Helper function that performs the actual write logic
+// Must be called within event loop context on the primary OSD
+int PGBackendTestFixture::do_create_and_write_impl(
const std::string& obj_name,
const std::string& data)
{
pg_t->create(hoid);
// Use persistent OBC so attr_cache is maintained across operations
- ObjectContextRef obc = get_or_create_obc(hoid, false, 0);
+ ObjectContextRef obc = get_object_context(hoid, true);
pg_t->obc_map[hoid] = obc;
// Note: We do NOT pre-seed attr_cache here. For a new object, attr_cache
bufferlist bl;
bl.append(data);
- pg_t->write(hoid, 0, bl.length(), bl);
+
+ // Only perform write if data is non-empty (PGTransaction requires len > 0)
+ if (bl.length() > 0) {
+ pg_t->write(hoid, 0, bl.length(), bl);
+ }
object_stat_sum_t delta_stats;
delta_stats.num_objects = 1;
return result;
}
+// Public interface that schedules the write on the primary OSD
+int PGBackendTestFixture::create_and_write(
+ const std::string& obj_name,
+ const std::string& data)
+{
+ // Get the primary OSD from the OSDMap
+ int primary_osd = osdmap->get_pg_acting_primary(pgid);
+ ceph_assert(primary_osd >= 0);
+
+ int result = -1;
+ event_loop->schedule_transaction(primary_osd, [this, &result, obj_name, data]() {
+ result = do_create_and_write_impl(obj_name, data);
+ });
+ event_loop->run_until_idle();
+
+ return result;
+}
+
+void PGBackendTestFixture::set_object_context(
+ const hobject_t& hoid,
+ ObjectContextRef obc)
+{
+ int osd = event_loop->get_current_executing_osd();
+ ceph_assert(osd != -1);
+ object_contexts[osd][hoid] = obc;
+}
+
+void PGBackendTestFixture::clear_object_contexts()
+{
+ int osd = event_loop->get_current_executing_osd();
+ ceph_assert(osd != -1);
+ object_contexts[osd].clear();
+}
+
ObjectContextRef PGBackendTestFixture::get_object_context(
- const hobject_t& hoid)
+ const hobject_t& hoid,
+ bool can_create,
+ const std::map<std::string, ceph::buffer::list, std::less<>> *attrs)
{
- PGBackend* primary_backend = get_primary_backend();
- ObjectContextRef obc = std::make_shared<ObjectContext>();
- obc->obs.oi = object_info_t(hoid);
- obc->obs.exists = false;
- obc->ssc = nullptr;
+ int osd = event_loop->get_current_executing_osd();
+ ceph_assert(osd != -1);
- // Try to read the ObjectInfo from the store
- ghobject_t ghoid(hoid, ghobject_t::NO_GEN, primary_backend->get_parent()->whoami_shard().shard);
- ceph::buffer::ptr value_ptr;
- int r = store->getattr(ch, ghoid, OI_ATTR, value_ptr);
- ceph_assert(r >= 0 && value_ptr.length() > 0);
+ // Check cache first (matches PrimaryLogPG::get_object_context line 11968)
+ auto it = object_contexts[osd].find(hoid);
+ if (it != object_contexts[osd].end()) {
+ return it->second;
+ }
- bufferlist bl;
- bl.append(value_ptr);
- auto p = bl.cbegin();
- obc->obs.oi.decode(p);
+ // Check disk for object info (matches PrimaryLogPG lines 11977-11982)
+ bufferlist bv;
+ if (attrs) {
+ auto it_oi = attrs->find(OI_ATTR);
+ ceph_assert(it_oi != attrs->end());
+ bv = it_oi->second;
+ } else {
+ PGBackend* primary_backend = get_primary_backend();
+ int r = primary_backend->objects_get_attr(hoid, OI_ATTR, &bv);
+
+ if (r < 0) {
+ if (!can_create) {
+ // Object doesn't exist and can't create (matches PrimaryLogPG lines 11985-11989)
+ return ObjectContextRef();
+ }
+
+ // Create new object context (matches PrimaryLogPG lines 11992-12004)
+ object_info_t oi(hoid);
+ ObjectContextRef obc = std::make_shared<ObjectContext>();
+ obc->obs.oi = oi;
+ obc->obs.exists = false;
+ obc->ssc = nullptr;
+ set_object_context(hoid, obc);
+ return obc;
+ }
+ }
+
+ // Decode object_info (matches PrimaryLogPG lines 12008-12015)
+ object_info_t oi;
+ try {
+ bufferlist::const_iterator bliter = bv.begin();
+ decode(oi, bliter);
+ } catch (...) {
+ return ObjectContextRef();
+ }
+
+ // Create OBC and populate from disk (matches PrimaryLogPG lines 12019-12022)
+ ObjectContextRef obc = std::make_shared<ObjectContext>();
+ obc->obs.oi = oi;
obc->obs.exists = true;
+ obc->ssc = nullptr;
+
+ // For EC pools, load all attributes (matches PrimaryLogPG lines 12031-12040)
+ if (pool_type == EC) {
+ if (attrs) {
+ obc->attr_cache = *attrs;
+ } else {
+ PGBackend* primary_backend = get_primary_backend();
+ int r = primary_backend->objects_get_attrs(hoid, &obc->attr_cache);
+ ceph_assert(r == 0);
+ }
+ }
+
+ // Cache the OBC (matches PrimaryLogPG line 12019 lookup_or_create)
+ set_object_context(hoid, obc);
return obc;
}
-int PGBackendTestFixture::write(
+// Helper function for write implementation
+int PGBackendTestFixture::do_write_impl(
const std::string& obj_name,
uint64_t offset,
const std::string& data,
hobject_t hoid = make_test_object(obj_name);
PGTransactionUPtr pg_t = std::make_unique<PGTransaction>();
- ObjectContextRef obc = get_or_create_obc(hoid, true, object_size);
+ ObjectContextRef obc = get_object_context(hoid, false);
pg_t->obc_map[hoid] = obc;
// Track outstanding write
return result;
}
+// Public interface that schedules the write on the primary OSD
+int PGBackendTestFixture::write(
+ const std::string& obj_name,
+ uint64_t offset,
+ const std::string& data,
+ uint64_t object_size)
+{
+ // Get the primary OSD from the OSDMap
+ int primary_osd = osdmap->get_pg_acting_primary(pgid);
+ ceph_assert(primary_osd >= 0);
+
+ int result = -1;
+ event_loop->schedule_transaction(primary_osd, [this, &result, obj_name, offset, data, object_size]() {
+ result = do_write_impl(obj_name, offset, data, object_size);
+ });
+ event_loop->run_until_idle();
+
+ return result;
+}
+
int PGBackendTestFixture::read_object(
const std::string& obj_name,
uint64_t offset,
}
}
- // Step 3: Clear all attr_caches before on_change()
- // The cached OI attributes may be stale after a peering event.
- // Also drop any stale outstanding write tracking: once we enter a new
- // interval, blocked/in-flight writes from the previous interval should no
- // longer prevent OBC reloading for rollback/recovery verification.
- clear_all_attr_caches();
+ // Step 3: Clear outstanding writes. FIXME: This belongs in the pg
outstanding_writes.clear();
// Step 4: Schedule on_change() calls as event loop actions
for (auto& [instance, be] : backends) {
if (be) {
PGBackend* backend_ptr = be.get();
- event_loop->schedule_peering_event(instance, [backend_ptr]() {
+ event_loop->schedule_peering_event(instance, [this, backend_ptr]() {
backend_ptr->on_change();
+ clear_object_contexts();
});
}
}
}
}
-void PGBackendTestFixture::clear_all_attr_caches()
-{
- // Clear attr_cache for all objects. This is called on on_change() to
- // invalidate cached attributes that might be stale after a peering event.
- for (auto& [hoid, obc] : object_contexts) {
- if (obc) {
- obc->attr_cache.clear();
- }
- }
-}
-
-
-int PGBackendTestFixture::write_attribute(
+// Helper function for write_attribute implementation
+int PGBackendTestFixture::do_write_attribute_impl(
const std::string& obj_name,
const std::string& attr_name,
const std::string& attr_value,
hobject_t hoid = make_test_object(obj_name);
PGTransactionUPtr pg_t = std::make_unique<PGTransaction>();
- ObjectContextRef obc = get_or_create_obc(hoid, true, 0);
+ ObjectContextRef obc = get_object_context(hoid, false);
pg_t->obc_map[hoid] = obc;
outstanding_writes[hoid]++;
return result;
}
+// Public interface that schedules the write on the primary OSD
+int PGBackendTestFixture::write_attribute(
+ const std::string& obj_name,
+ const std::string& attr_name,
+ const std::string& attr_value,
+ bool force_all_shards)
+{
+ // Get the primary OSD from the OSDMap
+ int primary_osd = osdmap->get_pg_acting_primary(pgid);
+ ceph_assert(primary_osd >= 0);
+
+ int result = -1;
+ event_loop->schedule_transaction(primary_osd, [this, &result, obj_name, attr_name, attr_value, force_all_shards]() {
+ result = do_write_attribute_impl(obj_name, attr_name, attr_value, force_all_shards);
+ });
+ event_loop->run_until_idle();
+
+ return result;
+}
+
object_info_t PGBackendTestFixture::read_shard_object_info(
const std::string& obj_name,
int shard)
/// Keyed by hobject_t, values are shared_ptr so the same OBC is reused
/// across sequential operations on the same object. This is critical for
/// EC attr_cache continuity.
- std::map<hobject_t, ObjectContextRef> object_contexts;
+ std::map<int, std::map<hobject_t, ObjectContextRef>> object_contexts;
/// Track outstanding writes per object. When this reaches 0, we can safely
/// clear attr_cache (as there are no in-flight writes that might have stale
obc->ssc = nullptr;
return obc;
}
-
- /// Get an existing OBC or create a new one.
- /// Unlike make_object_context(), this method reuses OBCs for the same
- /// object across operations, which is essential for attr_cache continuity
- /// in EC pools.
- /// @param primary_shard The shard ID to read attributes from (for EC pools)
- ObjectContextRef get_or_create_obc(
- const hobject_t& hoid,
- bool exists = false,
- uint64_t size = 0,
- int primary_shard = 0)
- {
- auto it = object_contexts.find(hoid);
- ObjectContextRef obc;
-
- if (it != object_contexts.end()) {
- obc = it->second;
- } else {
- obc = make_object_context(hoid, exists, size);
- object_contexts[hoid] = obc;
- }
-
- // If the object exists and this is an EC pool, populate attr_cache with
- // ALL attributes from disk if not already populated. This matches production
- // behavior where the OBC is loaded with all xattrs from the object store.
- // In EC, attributes are stored per-shard, so we must read from the specified shard.
- if (exists && pool_type == EC && store && !chs.empty() && obc->attr_cache.empty()) {
- auto writes_it = outstanding_writes.find(hoid);
- bool has_outstanding_writes = (writes_it != outstanding_writes.end() && writes_it->second > 0);
-
- // Cannot read from disk if there are outstanding writes - test bug
- ceph_assert(!has_outstanding_writes);
-
- // For EC pools, attributes are stored with the shard ID in the ghobject_t
- ceph_assert(primary_shard >= 0 && primary_shard < (int)chs.size());
- ObjectStore::CollectionHandle ch = chs[primary_shard];
- if (ch) {
- ghobject_t ghoid(hoid, ghobject_t::NO_GEN, shard_id_t(primary_shard));
- std::map<std::string, ceph::buffer::ptr, std::less<>> attrs;
- int r = store->getattrs(ch, ghoid, attrs);
-
- if (r >= 0) {
- // Successfully read all attributes from disk - populate the cache
- for (auto& [key, value_ptr] : attrs) {
- bufferlist bl;
- bl.append(value_ptr);
- obc->attr_cache[key] = std::move(bl);
- }
- }
- }
- }
-
- return obc;
- }
-
+
/**
* Set the next version number for auto-generation.
* This can be used by tests after rollback to set the version to a specific value.
}
/**
- * Read ObjectInfo from the store for an existing object.
- * Returns an ObjectContext with the decoded ObjectInfo, or a new
- * ObjectContext with default values if the object doesn't exist.
+ * Set an object context for the given object.
+ * This encapsulates access to the per-OSD object_contexts map.
+ * Must be called within event loop context.
+ *
+ * @param hoid The object to set context for
+ * @param obc The object context to cache
+ */
+ void set_object_context(
+ const hobject_t& hoid,
+ ObjectContextRef obc);
+
+ /**
+ * Clear all object contexts for the current OSD.
+ * This encapsulates access to the per-OSD object_contexts map.
+ * Must be called within event loop context.
+ */
+ void clear_object_contexts();
+
+ /**
+ * Get or create an object context for the given object.
+ * This matches PrimaryLogPG::get_object_context behavior.
+ *
+ * @param hoid The object to get context for
+ * @param can_create If true, create a new OBC if object doesn't exist
+ * @param attrs Optional attributes to use instead of reading from disk
*/
ObjectContextRef get_object_context(
- const hobject_t& hoid);
+ const hobject_t& hoid,
+ bool can_create,
+ const std::map<std::string, ceph::buffer::list, std::less<>> *attrs = nullptr);
int do_transaction_and_complete(
const hobject_t& hoid,
std::vector<pg_log_entry_t> log_entries,
std::function<void(int)> on_write_complete = nullptr);
+ // Helper functions that perform the actual write logic
+ // Must be called within event loop context on the primary OSD
+ int do_create_and_write_impl(
+ const std::string& obj_name,
+ const std::string& data);
+
+ int do_write_impl(
+ const std::string& obj_name,
+ uint64_t offset,
+ const std::string& data,
+ uint64_t object_size);
+
+ int do_write_attribute_impl(
+ const std::string& obj_name,
+ const std::string& attr_name,
+ const std::string& attr_value,
+ bool force_all_shards);
+
virtual int create_and_write(
const std::string& obj_name,
const std::string& data);
std::shared_ptr<OSDMap> new_osdmap,
std::optional<pg_shard_t> new_primary = std::nullopt);
- /**
- * Clear attr_cache for all objects.
- * Called on on_change() to invalidate cached attributes that might be stale
- * after a peering event or OSDMap change.
- */
- void clear_all_attr_caches();
-
/**
* Write attributes to an object with control over first_write_in_interval.
*
<< "on_activate_complete should have been called during peering";
}
+TEST_P(TestECFailoverWithPeering, ZeroSizeObjectWithAttributesRecovery) {
+ // ASSERT_TRUE(all_shards_active()) << "Initial peering must complete";
+
+ const std::string obj_name = "test_primary_failover";
+ const std::string test_data;
+
+ create_and_write(obj_name, test_data);
+
+ // Mark OSD 0 (the initial primary) as down
+ // PeeringState will automatically determine the new primary
+ mark_osd_down(0);
+
+ write_attribute(obj_name, "key", "value", false);
+
+ // Determine the actual new primary from the OSDMap
+ int new_primary_shard = get_primary_shard_from_osdmap();
+ ASSERT_GE(new_primary_shard, 0) << "Should have a valid new primary after failover";
+
+ // For an optimized EC pool (k=4, m=2), the new primary should be a coding shard (>= k)
+ // For a non-optimized pool, it would be shard 1
+ const pg_pool_t& pool = get_pool();
+ if (pool.allows_ecoptimizations()) {
+ ASSERT_GE(new_primary_shard, k)
+ << "New primary should be a coding shard (>= k) for optimized pool";
+ } else {
+ ASSERT_EQ(new_primary_shard, 1)
+ << "New primary should be shard 1 for non-optimized pool";
+ }
+
+ ASSERT_TRUE(get_peering_listener(new_primary_shard)->backend_listener->pgb_is_primary())
+ << "Shard " << new_primary_shard << " should be new primary";
+
+ ASSERT_FALSE(get_peering_listener(0)->backend_listener->pgb_is_primary())
+ << "Failed shard should not be primary";
+
+ std::string state = get_state_name(new_primary_shard);
+ ASSERT_TRUE(state.find("Active") != std::string::npos)
+ << "New primary should be Active after failover, got: " << state;
+
+ // Verify the PG reached Active state
+ ASSERT_TRUE(get_peering_state(new_primary_shard)->is_active())
+ << "New primary should be in Active state";
+
+ mark_osd_up(0);
+
+ run_recovery_and_verify_callbacks(obj_name, 0, test_data);
+
+ // Verify that the attribute was recovered on shard 0
+ hobject_t hoid = make_test_object(obj_name);
+ ghobject_t ghoid = ghobject_t(hoid, ghobject_t::NO_GEN, shard_id_t(0));
+
+ ceph::buffer::ptr attr_value;
+ int r = store->getattr(chs[0], ghoid, "key", attr_value);
+ ASSERT_GE(r, 0) << "Attribute 'key' should exist on recovered shard 0";
+ ASSERT_EQ(std::string(attr_value.c_str(), attr_value.length()), "value")
+ << "Attribute 'key' should have value 'value' after recovery";
+}
+
// ---------------------------------------------------------------------------
// EC backend configurations for parameterized tests
// ---------------------------------------------------------------------------
projects / ceph.git / commitdiff