read_request_t &read_request,
const bool for_recovery,
bool want_attrs,
- bool want_omap_header) {
- if (want_omap_header) {
+ bool want_omap_header,
+ bool want_omap_keys) {
+ if (want_omap_header || want_omap_keys) {
ceph_assert(get_parent()->get_pool().supports_omap());
}
return -EIO;
}
- bool need_attr_request = want_attrs || want_omap_header;
+ bool need_attr_request = want_attrs || want_omap_header || want_omap_keys;
read_request.want_attrs = want_attrs;
read_request.want_omap_header = want_omap_header;
+ read_request.want_omap_keys = want_omap_keys;
// Rather than repeating whole read, we can remove everything we already have.
for (auto iter = read_request.shard_reads.begin();
}
if (need_attr_request) {
- // This happens if we got an error from the shard where we were requesting
- // the attributes from and the recovery does not require any non primary
- // shards. The example seen in test was a 2+1 EC being recovered. shards 0
- // and 2 were being requested and read as part of recovery. Shard was reading
- // the attributes and failed. The recovery required shard 1, but that does
- // not have valid attributes on it, so the attribute read failed.
- // This is a pretty obscure case, so no need to optimise that much. Do an
- // empty read!
- shard_id_set have;
- shard_id_map<pg_shard_t> pg_shards(sinfo.get_k_plus_m());
- get_all_avail_shards(hoid, have, pg_shards, for_recovery, error_shards);
- for (auto shard : have) {
- if (!sinfo.is_nonprimary_shard(shard)) {
- shard_read_t shard_read;
- shard_read.pg_shard = pg_shards[shard];
- read_request.shard_reads.insert(shard, shard_read_t());
- break;
- }
+ int r = ensure_primary_shard_for_omap(hoid, read_request, for_recovery, error_shards);
+ if (r != 0) {
+ return r;
}
}
+ if (read_request.shard_reads.empty()) {
+ dout(0) << __func__ << " no shards to read for " << hoid
+ << " after filtering already-read extents" << dendl;
+ return -EIO;
+ }
+
return 0;
}
+int ECCommon::ReadPipeline::ensure_primary_shard_for_omap(
+ const hobject_t &hoid,
+ read_request_t &read_request,
+ const bool for_recovery,
+ const std::optional<std::set<pg_shard_t>> &error_shards) {
+
+ if (!get_parent()->get_pool().supports_omap()) {
+ return 0;
+ }
+ if (!read_request.want_omap_header && !read_request.want_omap_keys) {
+ return 0;
+ }
+
+ // Check if there is already a suitable primary-capable shard in shard_reads
+ for (auto &[shard_id, shard_read] : read_request.shard_reads) {
+ if (!sinfo.is_nonprimary_shard(shard_id)) {
+ // Found a primary-capable shard, verify it has clean omap
+ if (for_recovery) {
+ const pg_missing_t &missing = get_parent()->get_shard_missing(shard_read.pg_shard);
+ auto miter = missing.get_items().find(hoid);
+ if (miter != missing.get_items().end() && miter->second.clean_regions.omap_is_dirty()) {
+ // This shard has dirty omap, keep looking
+ continue;
+ }
+ }
+ dout(20) << __func__ << ": existing shard " << shard_id
+ << " can handle omap read for " << hoid << dendl;
+ return 0;
+ }
+ }
+
+ // No suitable shard exists in shard_reads, need to add one
+ shard_id_set have;
+ shard_id_map<pg_shard_t> pg_shards(sinfo.get_k_plus_m());
+ get_all_avail_shards(hoid, have, pg_shards, for_recovery, error_shards);
+ for (auto shard : have) {
+ if (!sinfo.is_nonprimary_shard(shard)) {
+ // Check if this shard has clean omap
+ if (for_recovery) {
+ const pg_missing_t &missing = get_parent()->get_shard_missing(pg_shards[shard]);
+ auto miter = missing.get_items().find(hoid);
+ if (miter != missing.get_items().end() && miter->second.clean_regions.omap_is_dirty()) {
+ dout(20) << __func__ << ": skipping shard " << shard
+ << " for " << hoid << " due to dirty omap" << dendl;
+ continue;
+ }
+ }
+
+ // Found a suitable shard, add it to shard_reads
+ shard_read_t shard_read;
+ shard_read.pg_shard = pg_shards[shard];
+ read_request.shard_reads.insert(shard, shard_read);
+ dout(20) << __func__ << ": added shard " << shard
+ << " for omap read of " << hoid << dendl;
+ return 0;
+ }
+ }
+
+ // No suitable shard found
+ dout(0) << __func__ << " no primary-capable shard with clean omap available for "
+ << hoid << dendl;
+ return -EIO;
+}
+
void ECCommon::ReadPipeline::start_read_op(
const int priority,
map<hobject_t, read_request_t> &to_read,
bool need_attrs = read_request.want_attrs;
bool need_omap_header = read_request.want_omap_header;
bool need_omap_keys = read_request.want_omap_keys;
- if (need_omap_header || need_omap_keys) {
+ bool need_omap = need_omap_header || need_omap_keys;
+ if (need_omap) {
ceph_assert(get_parent()->get_pool().supports_omap());
}
for (auto &&[shard, shard_read]: read_request.shard_reads) {
if (need_attrs && !sinfo.is_nonprimary_shard(shard)) {
messages[shard_read.pg_shard].attrs_to_read.insert(hoid);
+ reads_sent = true;
need_attrs = false;
}
- if (need_omap_header && !sinfo.is_nonprimary_shard(shard)) {
- messages[shard_read.pg_shard].omap_headers_to_read.insert(hoid);
+ if (need_omap && !sinfo.is_nonprimary_shard(shard)) {
+ if (need_omap_header) {
+ messages[shard_read.pg_shard].omap_headers_to_read.insert(hoid);
+ reads_sent = true;
+ }
+ if (need_omap_keys) {
+ messages[shard_read.pg_shard].omap_read_from.insert(
+ {hoid, {read_request.omap_read_from, read_request.omap_max_bytes}});
+ reads_sent = true;
+ }
+ shard_read.omap_source = true;
+ need_omap = false;
need_omap_header = false;
- }
- if (need_omap_keys && !sinfo.is_nonprimary_shard(shard)) {
- messages[shard_read.pg_shard].omap_read_from.insert(
- {hoid, {read_request.omap_read_from, read_request.omap_max_bytes}});
need_omap_keys = false;
}
if (shard_read.subchunk) {
messages[shard_read.pg_shard].subchunks[hoid] = *shard_read.subchunk;
+ reads_sent = true;
} else {
static const std::vector default_sub_chunk = {make_pair(0, 1)};
messages[shard_read.pg_shard].subchunks[hoid] = default_sub_chunk;
+ reads_sent = true;
}
rop.obj_to_source[hoid].insert(shard_read.pg_shard);
rop.source_to_obj[shard_read.pg_shard].insert(hoid);
}
for (auto &[_, shard_read]: read_request.shard_reads) {
- ceph_assert(!shard_read.extents.empty());
rop.debug_log.emplace_back(ECUtil::READ_REQUEST, shard_read.pg_shard,
shard_read.extents);
for (auto &[start, len]: shard_read.extents) {
}
}
ceph_assert(!need_attrs);
+ ceph_assert(!need_omap);
ceph_assert(!need_omap_header);
ceph_assert(!need_omap_keys);
}
dout(10) << __func__ << " want omap_header again" << dendl;
}
+ // Check if we need to read omap_keys again
+ const bool want_omap_keys =
+ rop.to_read.at(hoid).want_omap_keys &&
+ (!rop.complete.at(hoid).omap_entries || rop.complete.at(hoid).omap_entries->empty());
+ if (want_omap_keys) {
+ ceph_assert(get_parent()->get_pool().supports_omap());
+ dout(10) << __func__ << " want omap_keys again" << dendl;
+ }
+
read_request_t &read_request = rop.to_read.at(hoid);
// reset the old shard reads, we are going to read them again.
read_request.shard_reads.clear();
return get_remaining_shards(hoid, rop.complete.at(hoid), read_request,
- rop.for_recovery, want_attrs, want_omap_header);
+ rop.for_recovery, want_attrs, want_omap_header,
+ want_omap_keys);
}
void ECCommon::ReadPipeline::kick_reads() {
bool ECCommon::shard_read_t::operator==(const shard_read_t &other) const {
return extents == other.extents &&
subchunk == other.subchunk &&
- pg_shard == other.pg_shard;
+ pg_shard == other.pg_shard &&
+ omap_source == other.omap_source;
}
bool ECCommon::read_request_t::operator==(const read_request_t &other) const {
recovery_ops.erase(op.hoid);
return;
}
- if (get_parent()->get_pool().supports_omap()) {
- shard_id_set have;
- shard_id_map<pg_shard_t> pg_shards(sinfo.get_k_plus_m());
- read_pipeline.get_all_avail_shards(op.hoid, have, pg_shards, true, {});
- bool found_omap_shard = false;
- for (const auto shard : have) {
- if (!sinfo.is_nonprimary_shard(shard)) {
- const pg_missing_t &missing = get_parent()->get_shard_missing(pg_shards[shard]);
- auto miter = missing.get_items().find(op.hoid);
- if (miter != missing.get_items().end() && miter->second.clean_regions.omap_is_dirty()) {
- dout(20) << __func__ << ": skipping shard " << shard
- << " for " << op.hoid << " due to dirty omap" << dendl;
- continue;
- }
- shard_read_t shard_read;
- shard_read.pg_shard = pg_shards[shard];
- read_request.shard_reads.insert(shard, shard_read);
- found_omap_shard = true;
- dout(10) << __func__ << ": selected shard " << shard
- << " for omap read of " << op.hoid << dendl;
- break;
- }
- }
- if (!found_omap_shard) {
- dout(10) << __func__ << ": ERROR: no shard with clean omap found for "
- << op.hoid << ", canceling recovery" << dendl;
- get_parent()->cancel_pull(op.hoid);
- recovery_ops.erase(op.hoid);
- return;
- }
+ int r2 = read_pipeline.ensure_primary_shard_for_omap(
+ op.hoid, read_request, true, {});
+ if (r2 != 0) {
+ get_parent()->cancel_pull(op.hoid);
+ recovery_ops.erase(op.hoid);
+ return;
}
if (read_request.shard_reads.empty()) {
ceph_assert(op.obc);
ASSERT_TRUE(returned_keys.empty());
}
+TEST_P(OmapTest, OmapPreservedAfterTruncateToZero) {
+ SKIP_IF_CRIMSON();
+
+ const std::string oid = "test_omap_truncate_to_zero";
+ const std::string omap_header = "test_header_for_truncate";
+ bufferlist omap_header_bl;
+ encode(omap_header, omap_header_bl);
+
+ // Create object with data and omap
+ auto omap_map = get_test_omap_data();
+ create_test_object_with_omap(oid, omap_map, omap_header_bl);
+
+ // Verify initial state - object has data
+ bufferlist read_bl;
+ int ret = ioctx.read(oid, read_bl, 0, 0);
+ EXPECT_GT(read_bl.length(), 0); // Object has data
+
+ // Truncate object to zero
+ ObjectWriteOperation write_op;
+ write_op.truncate(0);
+ ret = ioctx.operate(oid, &write_op);
+ EXPECT_EQ(ret, 0);
+
+ // Verification 1: Object data is truncated to zero
+ read_bl.clear();
+ ret = ioctx.read(oid, read_bl, 0, 0);
+ EXPECT_EQ(ret, 0); // No data to read
+ EXPECT_EQ(read_bl.length(), 0); // Zero length
+
+ // Verification 2: Omap header is preserved
+ bufferlist result_header_bl;
+ int err = 0;
+ ObjectReadOperation read_op;
+ read_op.omap_get_header(&result_header_bl, &err);
+ ret = ioctx.operate(oid, &read_op, nullptr);
+ EXPECT_EQ(ret, 0);
+ EXPECT_EQ(err, 0);
+
+ // Check if header bufferlist has data before decoding
+ ASSERT_GT(result_header_bl.length(), 0u)
+ << "Expected omap header to be preserved after truncate to zero, but header is empty";
+
+ std::string result_header;
+ decode(result_header, result_header_bl);
+ EXPECT_EQ(result_header, omap_header)
+ << "Expected omap header '" << omap_header << "' but got '" << result_header << "'";
+
+ // Verification 3: All omap keys are preserved
+ std::map<std::string, bufferlist> vals_read;
+ err = 0;
+ ObjectReadOperation read_op2;
+ read_op2.omap_get_vals2("", LONG_MAX, &vals_read, nullptr, &err);
+ ret = ioctx.operate(oid, &read_op2, nullptr);
+ EXPECT_EQ(ret, 0);
+ EXPECT_EQ(err, 0);
+ EXPECT_EQ(vals_read.size(), omap_map.size())
+ << "Expected " << omap_map.size() << " omap keys to be preserved after truncate to zero, but got " << vals_read.size();
+
+ // Verify each key-value pair is preserved
+ for (const auto& [key, expected_bl] : omap_map) {
+ ASSERT_TRUE(vals_read.find(key) != vals_read.end())
+ << "Expected omap key '" << key << "' to be preserved after truncate to zero, but it is missing";
+
+ // Check if value bufferlist has data before decoding
+ ASSERT_GT(vals_read[key].length(), 0u)
+ << "Expected omap value for key '" << key << "' to be preserved, but value is empty";
+
+ std::string expected_val, actual_val;
+ decode(expected_val, expected_bl);
+ decode(actual_val, vals_read[key]);
+ EXPECT_EQ(actual_val, expected_val)
+ << "Expected omap value '" << expected_val << "' for key '" << key << "' but got '" << actual_val << "'";
+ }
+}
+
+
TEST_P(OmapTest, OmapRecovery) {
SKIP_IF_CRIMSON();
turn_balancing_off();
projects / ceph.git / commitdiff