diff options
Diffstat (limited to 'src/crimson/osd/replicated_recovery_backend.cc')
| -rw-r--r-- | src/crimson/osd/replicated_recovery_backend.cc | 1076 |
1 files changed, 1076 insertions, 0 deletions
diff --git a/src/crimson/osd/replicated_recovery_backend.cc b/src/crimson/osd/replicated_recovery_backend.cc new file mode 100644 index 000000000..0812003bb --- /dev/null +++ b/src/crimson/osd/replicated_recovery_backend.cc @@ -0,0 +1,1076 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <fmt/format.h> +#include <fmt/ostream.h> +#include <seastar/core/future.hh> +#include <seastar/core/do_with.hh> + +#include "crimson/osd/pg.h" +#include "crimson/osd/pg_backend.h" +#include "replicated_recovery_backend.h" + +#include "msg/Message.h" + +namespace { + seastar::logger& logger() { + return crimson::get_logger(ceph_subsys_osd); + } +} + +seastar::future<> ReplicatedRecoveryBackend::recover_object( + const hobject_t& soid, + eversion_t need) +{ + logger().debug("{}: {}, {}", __func__, soid, need); + // always add_recovering(soid) before recover_object(soid) + assert(is_recovering(soid)); + // start tracking the recovery of soid + return maybe_pull_missing_obj(soid, need).then([this, soid, need] { + logger().debug("recover_object: loading obc: {}", soid); + return pg.with_head_obc<RWState::RWREAD>(soid, + [this, soid, need](auto obc) { + logger().debug("recover_object: loaded obc: {}", obc->obs.oi.soid); + auto& recovery_waiter = recovering.at(soid); + recovery_waiter.obc = obc; + recovery_waiter.obc->wait_recovery_read(); + return maybe_push_shards(soid, need); + }).handle_error( + crimson::osd::PG::load_obc_ertr::all_same_way([soid](auto& code) { + // TODO: may need eio handling? + logger().error("recover_object saw error code {}, ignoring object {}", + code, soid); + })); + }); +} + +seastar::future<> +ReplicatedRecoveryBackend::maybe_push_shards( + const hobject_t& soid, + eversion_t need) +{ + return seastar::parallel_for_each(get_shards_to_push(soid), + [this, need, soid](auto shard) { + return prep_push(soid, need, shard).then([this, soid, shard](auto push) { + auto msg = make_message<MOSDPGPush>(); + msg->from = pg.get_pg_whoami(); + msg->pgid = pg.get_pgid(); + msg->map_epoch = pg.get_osdmap_epoch(); + msg->min_epoch = pg.get_last_peering_reset(); + msg->pushes.push_back(std::move(push)); + msg->set_priority(pg.get_recovery_op_priority()); + return shard_services.send_to_osd(shard.osd, + std::move(msg), + pg.get_osdmap_epoch()).then( + [this, soid, shard] { + return recovering.at(soid).wait_for_pushes(shard); + }); + }); + }).then([this, soid] { + auto &recovery = recovering.at(soid); + auto push_info = recovery.pushing.begin(); + object_stat_sum_t stat = {}; + if (push_info != recovery.pushing.end()) { + stat = push_info->second.stat; + } else { + // no push happened, take pull_info's stat + assert(recovery.pi); + stat = recovery.pi->stat; + } + pg.get_recovery_handler()->on_global_recover(soid, stat, false); + return seastar::make_ready_future<>(); + }).handle_exception([this, soid](auto e) { + auto &recovery = recovering.at(soid); + if (recovery.obc) { + recovery.obc->drop_recovery_read(); + } + recovering.erase(soid); + return seastar::make_exception_future<>(e); + }); +} + +seastar::future<> +ReplicatedRecoveryBackend::maybe_pull_missing_obj( + const hobject_t& soid, + eversion_t need) +{ + pg_missing_tracker_t local_missing = pg.get_local_missing(); + if (!local_missing.is_missing(soid)) { + return seastar::make_ready_future<>(); + } + PullOp po; + auto& recovery_waiter = recovering.at(soid); + recovery_waiter.pi = std::make_optional<RecoveryBackend::PullInfo>(); + auto& pi = *recovery_waiter.pi; + prepare_pull(po, pi, soid, need); + auto msg = make_message<MOSDPGPull>(); + msg->from = pg.get_pg_whoami(); + msg->set_priority(pg.get_recovery_op_priority()); + msg->pgid = pg.get_pgid(); + msg->map_epoch = pg.get_osdmap_epoch(); + msg->min_epoch = pg.get_last_peering_reset(); + msg->set_pulls({std::move(po)}); + return shard_services.send_to_osd( + pi.from.osd, + std::move(msg), + pg.get_osdmap_epoch() + ).then([&recovery_waiter] { + return recovery_waiter.wait_for_pull(); + }); +} + +seastar::future<> ReplicatedRecoveryBackend::push_delete( + const hobject_t& soid, + eversion_t need) +{ + logger().debug("{}: {}, {}", __func__, soid, need); + recovering[soid]; + epoch_t min_epoch = pg.get_last_peering_reset(); + + assert(pg.get_acting_recovery_backfill().size() > 0); + return seastar::parallel_for_each(pg.get_acting_recovery_backfill(), + [this, soid, need, min_epoch](pg_shard_t shard) { + if (shard == pg.get_pg_whoami()) + return seastar::make_ready_future<>(); + auto iter = pg.get_shard_missing().find(shard); + if (iter == pg.get_shard_missing().end()) + return seastar::make_ready_future<>(); + if (iter->second.is_missing(soid)) { + logger().debug("push_delete: will remove {} from {}", soid, shard); + pg.begin_peer_recover(shard, soid); + spg_t target_pg(pg.get_info().pgid.pgid, shard.shard); + auto msg = make_message<MOSDPGRecoveryDelete>( + pg.get_pg_whoami(), target_pg, pg.get_osdmap_epoch(), min_epoch); + msg->set_priority(pg.get_recovery_op_priority()); + msg->objects.push_back(std::make_pair(soid, need)); + return shard_services.send_to_osd(shard.osd, std::move(msg), + pg.get_osdmap_epoch()).then( + [this, soid, shard] { + return recovering.at(soid).wait_for_pushes(shard); + }); + } + return seastar::make_ready_future<>(); + }); +} + +seastar::future<> ReplicatedRecoveryBackend::handle_recovery_delete( + Ref<MOSDPGRecoveryDelete> m) +{ + logger().debug("{}: {}", __func__, *m); + + auto& p = m->objects.front(); //TODO: only one delete per message for now. + return local_recover_delete(p.first, p.second, pg.get_osdmap_epoch()).then( + [this, m] { + auto reply = make_message<MOSDPGRecoveryDeleteReply>(); + reply->from = pg.get_pg_whoami(); + reply->set_priority(m->get_priority()); + reply->pgid = spg_t(pg.get_info().pgid.pgid, m->from.shard); + reply->map_epoch = m->map_epoch; + reply->min_epoch = m->min_epoch; + reply->objects = m->objects; + return shard_services.send_to_osd(m->from.osd, std::move(reply), pg.get_osdmap_epoch()); + }); +} + +seastar::future<> ReplicatedRecoveryBackend::on_local_recover_persist( + const hobject_t& soid, + const ObjectRecoveryInfo& _recovery_info, + bool is_delete, + epoch_t epoch_frozen) +{ + logger().debug("{}", __func__); + ceph::os::Transaction t; + pg.get_recovery_handler()->on_local_recover(soid, _recovery_info, is_delete, t); + return shard_services.get_store().do_transaction(coll, std::move(t)).then( + [this, epoch_frozen, last_complete = pg.get_info().last_complete] { + pg.get_recovery_handler()->_committed_pushed_object(epoch_frozen, last_complete); + return seastar::make_ready_future<>(); + }); +} + +seastar::future<> ReplicatedRecoveryBackend::local_recover_delete( + const hobject_t& soid, + eversion_t need, + epoch_t epoch_to_freeze) +{ + logger().debug("{}: {}, {}", __func__, soid, need); + return backend->load_metadata(soid).safe_then([this] + (auto lomt) { + if (lomt->os.exists) { + return seastar::do_with(ceph::os::Transaction(), + [this, lomt = std::move(lomt)](auto& txn) { + return backend->remove(lomt->os, txn).then([this, &txn]() mutable { + return shard_services.get_store().do_transaction(coll, + std::move(txn)); + }); + }); + } + return seastar::make_ready_future<>(); + }).safe_then([this, soid, epoch_to_freeze, need] { + ObjectRecoveryInfo recovery_info; + recovery_info.soid = soid; + recovery_info.version = need; + return on_local_recover_persist(soid, recovery_info, + true, epoch_to_freeze); + }, PGBackend::load_metadata_ertr::all_same_way( + [this, soid, epoch_to_freeze, need] (auto e) { + ObjectRecoveryInfo recovery_info; + recovery_info.soid = soid; + recovery_info.version = need; + return on_local_recover_persist(soid, recovery_info, + true, epoch_to_freeze); + }) + ); +} + +seastar::future<> ReplicatedRecoveryBackend::recover_delete( + const hobject_t &soid, eversion_t need) +{ + logger().debug("{}: {}, {}", __func__, soid, need); + + epoch_t cur_epoch = pg.get_osdmap_epoch(); + return seastar::do_with(object_stat_sum_t(), + [this, soid, need, cur_epoch](auto& stat_diff) { + return local_recover_delete(soid, need, cur_epoch).then( + [this, &stat_diff, cur_epoch, soid, need] { + if (!pg.has_reset_since(cur_epoch)) { + bool object_missing = false; + for (const auto& shard : pg.get_acting_recovery_backfill()) { + if (shard == pg.get_pg_whoami()) + continue; + if (pg.get_shard_missing(shard)->is_missing(soid)) { + logger().debug("recover_delete: soid {} needs to deleted from replca {}", + soid, shard); + object_missing = true; + break; + } + } + + if (!object_missing) { + stat_diff.num_objects_recovered = 1; + return seastar::make_ready_future<>(); + } else { + return push_delete(soid, need); + } + } + return seastar::make_ready_future<>(); + }).then([this, soid, &stat_diff] { + pg.get_recovery_handler()->on_global_recover(soid, stat_diff, true); + return seastar::make_ready_future<>(); + }); + }); +} + +seastar::future<PushOp> +ReplicatedRecoveryBackend::prep_push( + const hobject_t& soid, + eversion_t need, + pg_shard_t pg_shard) +{ + logger().debug("{}: {}, {}", __func__, soid, need); + + auto& recovery_waiter = recovering.at(soid); + auto& obc = recovery_waiter.obc; + interval_set<uint64_t> data_subset; + if (obc->obs.oi.size) { + data_subset.insert(0, obc->obs.oi.size); + } + const auto& missing = pg.get_shard_missing().find(pg_shard)->second; + if (HAVE_FEATURE(pg.min_peer_features(), SERVER_OCTOPUS)) { + const auto it = missing.get_items().find(soid); + assert(it != missing.get_items().end()); + data_subset.intersection_of(it->second.clean_regions.get_dirty_regions()); + logger().debug("prep_push: {} data_subset {}", soid, data_subset); + } + + logger().debug("prep_push: {} to {}", soid, pg_shard); + auto& pi = recovery_waiter.pushing[pg_shard]; + pg.begin_peer_recover(pg_shard, soid); + const auto pmissing_iter = pg.get_shard_missing().find(pg_shard); + const auto missing_iter = pmissing_iter->second.get_items().find(soid); + assert(missing_iter != pmissing_iter->second.get_items().end()); + + pi.obc = obc; + pi.recovery_info.size = obc->obs.oi.size; + pi.recovery_info.copy_subset = data_subset; + pi.recovery_info.soid = soid; + pi.recovery_info.oi = obc->obs.oi; + pi.recovery_info.version = obc->obs.oi.version; + pi.recovery_info.object_exist = + missing_iter->second.clean_regions.object_is_exist(); + pi.recovery_progress.omap_complete = + (!missing_iter->second.clean_regions.omap_is_dirty() && + HAVE_FEATURE(pg.min_peer_features(), SERVER_OCTOPUS)); + + return build_push_op(pi.recovery_info, pi.recovery_progress, &pi.stat).then( + [this, soid, pg_shard](auto pop) { + auto& recovery_waiter = recovering.at(soid); + auto& pi = recovery_waiter.pushing[pg_shard]; + pi.recovery_progress = pop.after_progress; + return pop; + }); +} + +void ReplicatedRecoveryBackend::prepare_pull(PullOp& po, PullInfo& pi, + const hobject_t& soid, + eversion_t need) { + logger().debug("{}: {}, {}", __func__, soid, need); + + pg_missing_tracker_t local_missing = pg.get_local_missing(); + const auto missing_iter = local_missing.get_items().find(soid); + auto m = pg.get_missing_loc_shards(); + pg_shard_t fromshard = *(m[soid].begin()); + + //TODO: skipped snap objects case for now + po.recovery_info.copy_subset.insert(0, (uint64_t) -1); + if (HAVE_FEATURE(pg.min_peer_features(), SERVER_OCTOPUS)) + po.recovery_info.copy_subset.intersection_of( + missing_iter->second.clean_regions.get_dirty_regions()); + po.recovery_info.size = ((uint64_t) -1); + po.recovery_info.object_exist = + missing_iter->second.clean_regions.object_is_exist(); + po.recovery_info.soid = soid; + po.soid = soid; + po.recovery_progress.data_complete = false; + po.recovery_progress.omap_complete = + !missing_iter->second.clean_regions.omap_is_dirty() && + HAVE_FEATURE(pg.min_peer_features(), SERVER_OCTOPUS); + po.recovery_progress.data_recovered_to = 0; + po.recovery_progress.first = true; + + pi.from = fromshard; + pi.soid = soid; + pi.recovery_info = po.recovery_info; + pi.recovery_progress = po.recovery_progress; +} + +seastar::future<PushOp> ReplicatedRecoveryBackend::build_push_op( + const ObjectRecoveryInfo& recovery_info, + const ObjectRecoveryProgress& progress, + object_stat_sum_t* stat) +{ + logger().debug("{} {} @{}", + __func__, recovery_info.soid, recovery_info.version); + return seastar::do_with(ObjectRecoveryProgress(progress), + uint64_t(crimson::common::local_conf() + ->osd_recovery_max_chunk), + recovery_info.version, + PushOp(), + [this, &recovery_info, &progress, stat] + (auto new_progress, auto available, auto v, auto pop) { + return read_metadata_for_push_op(recovery_info.soid, + progress, new_progress, + v, &pop).then([&](eversion_t local_ver) mutable { + // If requestor didn't know the version, use ours + if (v == eversion_t()) { + v = local_ver; + } else if (v != local_ver) { + logger().error("build_push_op: {} push {} v{} failed because local copy is {}", + pg.get_pgid(), recovery_info.soid, recovery_info.version, local_ver); + // TODO: bail out + } + return read_omap_for_push_op(recovery_info.soid, + progress, + new_progress, + &available, &pop); + }).then([this, &recovery_info, &progress, &available, &pop]() mutable { + logger().debug("build_push_op: available: {}, copy_subset: {}", + available, recovery_info.copy_subset); + return read_object_for_push_op(recovery_info.soid, + recovery_info.copy_subset, + progress.data_recovered_to, + available, &pop); + }).then([&recovery_info, &v, &progress, &new_progress, stat, &pop] + (uint64_t recovered_to) mutable { + new_progress.data_recovered_to = recovered_to; + if (new_progress.is_complete(recovery_info)) { + new_progress.data_complete = true; + if (stat) + stat->num_objects_recovered++; + } else if (progress.first && progress.omap_complete) { + // If omap is not changed, we need recovery omap + // when recovery cannot be completed once + new_progress.omap_complete = false; + } + if (stat) { + stat->num_keys_recovered += pop.omap_entries.size(); + stat->num_bytes_recovered += pop.data.length(); + } + pop.version = v; + pop.soid = recovery_info.soid; + pop.recovery_info = recovery_info; + pop.after_progress = new_progress; + pop.before_progress = progress; + logger().debug("build_push_op: pop version: {}, pop data length: {}", + pop.version, pop.data.length()); + return seastar::make_ready_future<PushOp>(std::move(pop)); + }); + }); +} + +seastar::future<eversion_t> +ReplicatedRecoveryBackend::read_metadata_for_push_op( + const hobject_t& oid, + const ObjectRecoveryProgress& progress, + ObjectRecoveryProgress& new_progress, + eversion_t ver, + PushOp* push_op) +{ + if (!progress.first) { + return seastar::make_ready_future<eversion_t>(ver); + } + return seastar::when_all_succeed( + backend->omap_get_header(coll, ghobject_t(oid)).handle_error( + crimson::os::FuturizedStore::read_errorator::all_same_way( + [] (const std::error_code& e) { + return seastar::make_ready_future<bufferlist>(); + })), + store->get_attrs(coll, ghobject_t(oid)).handle_error( + crimson::os::FuturizedStore::get_attrs_ertr::all_same_way( + [] (const std::error_code& e) { + return seastar::make_ready_future<crimson::os::FuturizedStore::attrs_t>(); + })) + ).then_unpack([&new_progress, push_op](auto bl, auto attrs) { + if (bl.length() == 0) { + logger().error("read_metadata_for_push_op: fail to read omap header"); + return eversion_t{}; + } else if (attrs.empty()) { + logger().error("read_metadata_for_push_op: fail to read attrs"); + return eversion_t{}; + } + push_op->omap_header.claim_append(std::move(bl)); + for (auto&& [key, val] : std::move(attrs)) { + push_op->attrset[key].push_back(val); + } + logger().debug("read_metadata_for_push_op: {}", push_op->attrset[OI_ATTR]); + object_info_t oi; + oi.decode(push_op->attrset[OI_ATTR]); + new_progress.first = false; + return oi.version; + }); +} + +seastar::future<uint64_t> +ReplicatedRecoveryBackend::read_object_for_push_op( + const hobject_t& oid, + const interval_set<uint64_t>& copy_subset, + uint64_t offset, + uint64_t max_len, + PushOp* push_op) +{ + if (max_len == 0 || copy_subset.empty()) { + push_op->data_included.clear(); + return seastar::make_ready_future<uint64_t>(offset); + } + // 1. get the extents in the interested range + return backend->fiemap(coll, ghobject_t{oid}, + 0, copy_subset.range_end()).then_wrapped( + [=](auto&& fiemap_included) mutable { + interval_set<uint64_t> extents; + try { + extents.intersection_of(copy_subset, fiemap_included.get0()); + } catch (std::exception &) { + // if fiemap() fails, we will read nothing, as the intersection of + // copy_subset and an empty interval_set would be empty anyway + extents.clear(); + } + // 2. we can read up to "max_len" bytes from "offset", so truncate the + // extents down to this quota. no need to return the number of consumed + // bytes, as this is the last consumer of this quota + push_op->data_included.span_of(extents, offset, max_len); + // 3. read the truncated extents + // TODO: check if the returned extents are pruned + return store->readv(coll, ghobject_t{oid}, push_op->data_included, 0); + }).safe_then([push_op, range_end=copy_subset.range_end()](auto &&bl) { + push_op->data.claim_append(std::move(bl)); + uint64_t recovered_to = 0; + if (push_op->data_included.empty()) { + // zero filled section, skip to end! + recovered_to = range_end; + } else { + // note down the progress, we will start from there next time + recovered_to = push_op->data_included.range_end(); + } + return seastar::make_ready_future<uint64_t>(recovered_to); + }, PGBackend::read_errorator::all_same_way([](auto e) { + logger().debug("build_push_op: read exception"); + return seastar::make_exception_future<uint64_t>(e); + })); +} + +seastar::future<> +ReplicatedRecoveryBackend::read_omap_for_push_op( + const hobject_t& oid, + const ObjectRecoveryProgress& progress, + ObjectRecoveryProgress& new_progress, + uint64_t* max_len, + PushOp* push_op) +{ + if (progress.omap_complete) { + return seastar::make_ready_future<>(); + } + return shard_services.get_store().get_omap_iterator(coll, ghobject_t{oid}) + .then([&progress, &new_progress, max_len, push_op](auto omap_iter) { + return omap_iter->lower_bound(progress.omap_recovered_to).then( + [omap_iter, &new_progress, max_len, push_op] { + return seastar::do_until([omap_iter, &new_progress, max_len, push_op] { + if (!omap_iter->valid()) { + new_progress.omap_complete = true; + return true; + } + if (push_op->omap_entries.empty()) { + return false; + } + if (const uint64_t entries_per_chunk = + crimson::common::local_conf()->osd_recovery_max_omap_entries_per_chunk; + entries_per_chunk > 0 && + push_op->omap_entries.size() >= entries_per_chunk) { + new_progress.omap_recovered_to = omap_iter->key(); + return true; + } + if (omap_iter->key().size() + omap_iter->value().length() > *max_len) { + new_progress.omap_recovered_to = omap_iter->key(); + return true; + } + return false; + }, + [omap_iter, max_len, push_op] { + push_op->omap_entries.emplace(omap_iter->key(), omap_iter->value()); + if (const uint64_t entry_size = + omap_iter->key().size() + omap_iter->value().length(); + entry_size > *max_len) { + *max_len -= entry_size; + } else { + *max_len = 0; + } + return omap_iter->next(); + }); + }); + }); +} + +std::vector<pg_shard_t> +ReplicatedRecoveryBackend::get_shards_to_push(const hobject_t& soid) const +{ + std::vector<pg_shard_t> shards; + assert(pg.get_acting_recovery_backfill().size() > 0); + for (const auto& peer : pg.get_acting_recovery_backfill()) { + if (peer == pg.get_pg_whoami()) + continue; + auto shard_missing = + pg.get_shard_missing().find(peer); + assert(shard_missing != pg.get_shard_missing().end()); + if (shard_missing->second.is_missing(soid)) { + shards.push_back(shard_missing->first); + } + } + return shards; +} + +seastar::future<> ReplicatedRecoveryBackend::handle_pull(Ref<MOSDPGPull> m) +{ + logger().debug("{}: {}", __func__, *m); + return seastar::parallel_for_each(m->take_pulls(), + [this, from=m->from](auto& pull_op) { + const hobject_t& soid = pull_op.soid; + logger().debug("handle_pull: {}", soid); + return backend->stat(coll, ghobject_t(soid)).then( + [this, &pull_op](auto st) { + ObjectRecoveryInfo &recovery_info = pull_op.recovery_info; + ObjectRecoveryProgress &progress = pull_op.recovery_progress; + if (progress.first && recovery_info.size == ((uint64_t) -1)) { + // Adjust size and copy_subset + recovery_info.size = st.st_size; + if (st.st_size) { + interval_set<uint64_t> object_range; + object_range.insert(0, st.st_size); + recovery_info.copy_subset.intersection_of(object_range); + } else { + recovery_info.copy_subset.clear(); + } + assert(recovery_info.clone_subset.empty()); + } + return build_push_op(recovery_info, progress, 0); + }).then([this, from](auto pop) { + auto msg = make_message<MOSDPGPush>(); + msg->from = pg.get_pg_whoami(); + msg->pgid = pg.get_pgid(); + msg->map_epoch = pg.get_osdmap_epoch(); + msg->min_epoch = pg.get_last_peering_reset(); + msg->set_priority(pg.get_recovery_op_priority()); + msg->pushes.push_back(std::move(pop)); + return shard_services.send_to_osd(from.osd, std::move(msg), + pg.get_osdmap_epoch()); + }); + }); +} + +seastar::future<bool> ReplicatedRecoveryBackend::_handle_pull_response( + pg_shard_t from, + const PushOp& pop, + PullOp* response, + ceph::os::Transaction* t) +{ + logger().debug("handle_pull_response {} {} data.size() is {} data_included: {}", + pop.recovery_info, pop.after_progress, pop.data.length(), pop.data_included); + + const hobject_t &hoid = pop.soid; + auto& recovery_waiter = recovering.at(hoid); + auto& pi = *recovery_waiter.pi; + if (pi.recovery_info.size == (uint64_t(-1))) { + pi.recovery_info.size = pop.recovery_info.size; + pi.recovery_info.copy_subset.intersection_of( + pop.recovery_info.copy_subset); + } + + // If primary doesn't have object info and didn't know version + if (pi.recovery_info.version == eversion_t()) + pi.recovery_info.version = pop.version; + + auto prepare_waiter = seastar::make_ready_future<>(); + if (pi.recovery_progress.first) { + prepare_waiter = pg.with_head_obc<RWState::RWNONE>( + pi.recovery_info.soid, [&pi, &recovery_waiter, &pop](auto obc) { + pi.obc = obc; + recovery_waiter.obc = obc; + obc->obs.oi.decode(pop.attrset.at(OI_ATTR)); + pi.recovery_info.oi = obc->obs.oi; + return crimson::osd::PG::load_obc_ertr::now(); + }).handle_error(crimson::ct_error::assert_all{}); + }; + return prepare_waiter.then([this, &pi, &pop, t, response]() mutable { + const bool first = pi.recovery_progress.first; + pi.recovery_progress = pop.after_progress; + logger().debug("new recovery_info {}, new progress {}", + pi.recovery_info, pi.recovery_progress); + interval_set<uint64_t> data_zeros; + { + uint64_t offset = pop.before_progress.data_recovered_to; + uint64_t length = (pop.after_progress.data_recovered_to - + pop.before_progress.data_recovered_to); + if (length) { + data_zeros.insert(offset, length); + } + } + auto [usable_intervals, data] = + trim_pushed_data(pi.recovery_info.copy_subset, + pop.data_included, pop.data); + bool complete = pi.is_complete(); + bool clear_omap = !pop.before_progress.omap_complete; + return submit_push_data(pi.recovery_info, first, complete, clear_omap, + std::move(data_zeros), usable_intervals, data, pop.omap_header, + pop.attrset, pop.omap_entries, t).then( + [this, response, &pi, &pop, complete, t, bytes_recovered=data.length()] { + pi.stat.num_keys_recovered += pop.omap_entries.size(); + pi.stat.num_bytes_recovered += bytes_recovered; + + if (complete) { + pi.stat.num_objects_recovered++; + pg.get_recovery_handler()->on_local_recover( + pop.soid, recovering.at(pop.soid).pi->recovery_info, + false, *t); + return true; + } else { + response->soid = pop.soid; + response->recovery_info = pi.recovery_info; + response->recovery_progress = pi.recovery_progress; + return false; + } + }); + }); +} + +seastar::future<> ReplicatedRecoveryBackend::handle_pull_response( + Ref<MOSDPGPush> m) +{ + const PushOp& pop = m->pushes[0]; //TODO: only one push per message for now. + if (pop.version == eversion_t()) { + // replica doesn't have it! + pg.get_recovery_handler()->on_failed_recover({ m->from }, pop.soid, + get_recovering(pop.soid).pi->recovery_info.version); + return seastar::make_exception_future<>( + std::runtime_error(fmt::format( + "Error on pushing side {} when pulling obj {}", + m->from, pop.soid))); + } + + logger().debug("{}: {}", __func__, *m); + return seastar::do_with(PullOp(), [this, m](auto& response) { + return seastar::do_with(ceph::os::Transaction(), m.get(), + [this, &response](auto& t, auto& m) { + pg_shard_t from = m->from; + PushOp& pop = m->pushes[0]; // only one push per message for now + return _handle_pull_response(from, pop, &response, &t).then( + [this, &t](bool complete) { + epoch_t epoch_frozen = pg.get_osdmap_epoch(); + return shard_services.get_store().do_transaction(coll, std::move(t)) + .then([this, epoch_frozen, complete, + last_complete = pg.get_info().last_complete] { + pg.get_recovery_handler()->_committed_pushed_object(epoch_frozen, last_complete); + return seastar::make_ready_future<bool>(complete); + }); + }); + }).then([this, m, &response](bool complete) { + if (complete) { + auto& pop = m->pushes[0]; + recovering.at(pop.soid).set_pulled(); + return seastar::make_ready_future<>(); + } else { + auto reply = make_message<MOSDPGPull>(); + reply->from = pg.get_pg_whoami(); + reply->set_priority(m->get_priority()); + reply->pgid = pg.get_info().pgid; + reply->map_epoch = m->map_epoch; + reply->min_epoch = m->min_epoch; + reply->set_pulls({std::move(response)}); + return shard_services.send_to_osd(m->from.osd, std::move(reply), pg.get_osdmap_epoch()); + } + }); + }); +} + +seastar::future<> ReplicatedRecoveryBackend::_handle_push( + pg_shard_t from, + const PushOp &pop, + PushReplyOp *response, + ceph::os::Transaction *t) +{ + logger().debug("{}", __func__); + + bool first = pop.before_progress.first; + interval_set<uint64_t> data_zeros; + { + uint64_t offset = pop.before_progress.data_recovered_to; + uint64_t length = (pop.after_progress.data_recovered_to - + pop.before_progress.data_recovered_to); + if (length) { + data_zeros.insert(offset, length); + } + } + bool complete = (pop.after_progress.data_complete && + pop.after_progress.omap_complete); + bool clear_omap = !pop.before_progress.omap_complete; + response->soid = pop.recovery_info.soid; + + return submit_push_data(pop.recovery_info, first, complete, clear_omap, + std::move(data_zeros), pop.data_included, pop.data, pop.omap_header, + pop.attrset, pop.omap_entries, t).then([this, complete, &pop, t] { + if (complete) { + pg.get_recovery_handler()->on_local_recover( + pop.recovery_info.soid, pop.recovery_info, + false, *t); + } + }); +} + +seastar::future<> ReplicatedRecoveryBackend::handle_push( + Ref<MOSDPGPush> m) +{ + if (pg.is_primary()) { + return handle_pull_response(m); + } + + logger().debug("{}: {}", __func__, *m); + return seastar::do_with(PushReplyOp(), [this, m](auto& response) { + const PushOp& pop = m->pushes[0]; //TODO: only one push per message for now + return seastar::do_with(ceph::os::Transaction(), + [this, m, &pop, &response](auto& t) { + return _handle_push(m->from, pop, &response, &t).then( + [this, &t] { + epoch_t epoch_frozen = pg.get_osdmap_epoch(); + return shard_services.get_store().do_transaction(coll, std::move(t)).then( + [this, epoch_frozen, last_complete = pg.get_info().last_complete] { + //TODO: this should be grouped with pg.on_local_recover somehow. + pg.get_recovery_handler()->_committed_pushed_object(epoch_frozen, last_complete); + }); + }); + }).then([this, m, &response]() mutable { + auto reply = make_message<MOSDPGPushReply>(); + reply->from = pg.get_pg_whoami(); + reply->set_priority(m->get_priority()); + reply->pgid = pg.get_info().pgid; + reply->map_epoch = m->map_epoch; + reply->min_epoch = m->min_epoch; + std::vector<PushReplyOp> replies = { std::move(response) }; + reply->replies.swap(replies); + return shard_services.send_to_osd(m->from.osd, + std::move(reply), pg.get_osdmap_epoch()); + }); + }); +} + +seastar::future<std::optional<PushOp>> +ReplicatedRecoveryBackend::_handle_push_reply( + pg_shard_t peer, + const PushReplyOp &op) +{ + const hobject_t& soid = op.soid; + logger().debug("{}, soid {}, from {}", __func__, soid, peer); + auto recovering_iter = recovering.find(soid); + if (recovering_iter == recovering.end() + || !recovering_iter->second.pushing.count(peer)) { + logger().debug("huh, i wasn't pushing {} to osd.{}", soid, peer); + return seastar::make_ready_future<std::optional<PushOp>>(); + } else { + auto& pi = recovering_iter->second.pushing[peer]; + bool error = pi.recovery_progress.error; + if (!pi.recovery_progress.data_complete && !error) { + return build_push_op(pi.recovery_info, pi.recovery_progress, + &pi.stat).then([&pi] (auto pop) { + pi.recovery_progress = pop.after_progress; + return seastar::make_ready_future<std::optional<PushOp>>(std::move(pop)); + }).handle_exception([recovering_iter, &pi, peer] (auto e) { + pi.recovery_progress.error = true; + recovering_iter->second.set_push_failed(peer, e); + return seastar::make_ready_future<std::optional<PushOp>>(); + }); + } + if (!error) { + pg.get_recovery_handler()->on_peer_recover(peer, soid, pi.recovery_info); + } + recovering_iter->second.set_pushed(peer); + return seastar::make_ready_future<std::optional<PushOp>>(); + } +} + +seastar::future<> ReplicatedRecoveryBackend::handle_push_reply( + Ref<MOSDPGPushReply> m) +{ + logger().debug("{}: {}", __func__, *m); + auto from = m->from; + auto& push_reply = m->replies[0]; //TODO: only one reply per message + + return _handle_push_reply(from, push_reply).then( + [this, from](std::optional<PushOp> push_op) { + if (push_op) { + auto msg = make_message<MOSDPGPush>(); + msg->from = pg.get_pg_whoami(); + msg->pgid = pg.get_pgid(); + msg->map_epoch = pg.get_osdmap_epoch(); + msg->min_epoch = pg.get_last_peering_reset(); + msg->set_priority(pg.get_recovery_op_priority()); + msg->pushes.push_back(std::move(*push_op)); + return shard_services.send_to_osd(from.osd, + std::move(msg), + pg.get_osdmap_epoch()); + } else { + return seastar::make_ready_future<>(); + } + }); +} + +std::pair<interval_set<uint64_t>, + bufferlist> +ReplicatedRecoveryBackend::trim_pushed_data( + const interval_set<uint64_t> ©_subset, + const interval_set<uint64_t> &intervals_received, + ceph::bufferlist data_received) +{ + logger().debug("{}", __func__); + // what i have is only a subset of what i want + if (intervals_received.subset_of(copy_subset)) { + return {intervals_received, data_received}; + } + // only collect the extents included by copy_subset and intervals_received + interval_set<uint64_t> intervals_usable; + bufferlist data_usable; + intervals_usable.intersection_of(copy_subset, intervals_received); + uint64_t have_off = 0; + for (auto [have_start, have_len] : intervals_received) { + interval_set<uint64_t> want; + want.insert(have_start, have_len); + want.intersection_of(copy_subset); + for (auto [want_start, want_len] : want) { + bufferlist sub; + uint64_t data_off = have_off + (want_start - have_start); + sub.substr_of(data_received, data_off, want_len); + data_usable.claim_append(sub); + } + have_off += have_len; + } + return {intervals_usable, data_usable}; +} + +seastar::future<> ReplicatedRecoveryBackend::submit_push_data( + const ObjectRecoveryInfo &recovery_info, + bool first, + bool complete, + bool clear_omap, + interval_set<uint64_t> data_zeros, + const interval_set<uint64_t> &intervals_included, + bufferlist data_included, + bufferlist omap_header, + const map<string, bufferlist> &attrs, + const map<string, bufferlist> &omap_entries, + ObjectStore::Transaction *t) +{ + logger().debug("{}", __func__); + hobject_t target_oid; + if (first && complete) { + target_oid = recovery_info.soid; + } else { + target_oid = get_temp_recovery_object(recovery_info.soid, + recovery_info.version); + if (first) { + logger().debug("{}: Adding oid {} in the temp collection", + __func__, target_oid); + add_temp_obj(target_oid); + } + } + + return [this, &recovery_info, first, complete, t, + &omap_header, &attrs, target_oid, clear_omap] { + if (first) { + if (!complete) { + t->remove(coll->get_cid(), ghobject_t(target_oid)); + t->touch(coll->get_cid(), ghobject_t(target_oid)); + bufferlist bv = attrs.at(OI_ATTR); + object_info_t oi(bv); + t->set_alloc_hint(coll->get_cid(), ghobject_t(target_oid), + oi.expected_object_size, + oi.expected_write_size, + oi.alloc_hint_flags); + } else { + if (!recovery_info.object_exist) { + t->remove(coll->get_cid(), ghobject_t(target_oid)); + t->touch(coll->get_cid(), ghobject_t(target_oid)); + bufferlist bv = attrs.at(OI_ATTR); + object_info_t oi(bv); + t->set_alloc_hint(coll->get_cid(), ghobject_t(target_oid), + oi.expected_object_size, + oi.expected_write_size, + oi.alloc_hint_flags); + } + //remove xattr and update later if overwrite on original object + t->rmattrs(coll->get_cid(), ghobject_t(target_oid)); + //if need update omap, clear the previous content first + if (clear_omap) + t->omap_clear(coll->get_cid(), ghobject_t(target_oid)); + } + + t->truncate(coll->get_cid(), ghobject_t(target_oid), recovery_info.size); + if (omap_header.length()) + t->omap_setheader(coll->get_cid(), ghobject_t(target_oid), omap_header); + + return store->stat(coll, ghobject_t(recovery_info.soid)).then( + [this, &recovery_info, complete, t, target_oid, + omap_header = std::move(omap_header)] (auto st) { + //TODO: pg num bytes counting + if (!complete) { + //clone overlap content in local object + if (recovery_info.object_exist) { + uint64_t local_size = std::min(recovery_info.size, (uint64_t)st.st_size); + interval_set<uint64_t> local_intervals_included, local_intervals_excluded; + if (local_size) { + local_intervals_included.insert(0, local_size); + local_intervals_excluded.intersection_of(local_intervals_included, recovery_info.copy_subset); + local_intervals_included.subtract(local_intervals_excluded); + } + for (auto [off, len] : local_intervals_included) { + logger().debug(" clone_range {} {}~{}", + recovery_info.soid, off, len); + t->clone_range(coll->get_cid(), ghobject_t(recovery_info.soid), + ghobject_t(target_oid), off, len, off); + } + } + } + return seastar::make_ready_future<>(); + }); + } + return seastar::make_ready_future<>(); + }().then([this, data_zeros=std::move(data_zeros), + &recovery_info, &intervals_included, t, target_oid, + &omap_entries, &attrs, data_included, complete, first]() mutable { + uint32_t fadvise_flags = CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL; + // Punch zeros for data, if fiemap indicates nothing but it is marked dirty + if (!data_zeros.empty()) { + data_zeros.intersection_of(recovery_info.copy_subset); + assert(intervals_included.subset_of(data_zeros)); + data_zeros.subtract(intervals_included); + + logger().debug("submit_push_data recovering object {} copy_subset: {} " + "intervals_included: {} data_zeros: {}", + recovery_info.soid, recovery_info.copy_subset, + intervals_included, data_zeros); + + for (auto [start, len] : data_zeros) { + t->zero(coll->get_cid(), ghobject_t(target_oid), start, len); + } + } + uint64_t off = 0; + for (auto [start, len] : intervals_included) { + bufferlist bit; + bit.substr_of(data_included, off, len); + t->write(coll->get_cid(), ghobject_t(target_oid), + start, len, bit, fadvise_flags); + off += len; + } + + if (!omap_entries.empty()) + t->omap_setkeys(coll->get_cid(), ghobject_t(target_oid), omap_entries); + if (!attrs.empty()) + t->setattrs(coll->get_cid(), ghobject_t(target_oid), attrs); + + if (complete) { + if (!first) { + logger().debug("submit_push_data: Removing oid {} from the temp collection", + target_oid); + clear_temp_obj(target_oid); + t->remove(coll->get_cid(), ghobject_t(recovery_info.soid)); + t->collection_move_rename(coll->get_cid(), ghobject_t(target_oid), + coll->get_cid(), ghobject_t(recovery_info.soid)); + } + submit_push_complete(recovery_info, t); + } + logger().debug("submit_push_data: done"); + return seastar::make_ready_future<>(); + }); +} + +void ReplicatedRecoveryBackend::submit_push_complete( + const ObjectRecoveryInfo &recovery_info, + ObjectStore::Transaction *t) +{ + for (const auto& [oid, extents] : recovery_info.clone_subset) { + for (const auto [off, len] : extents) { + logger().debug(" clone_range {} {}~{}", oid, off, len); + t->clone_range(coll->get_cid(), ghobject_t(oid), ghobject_t(recovery_info.soid), + off, len, off); + } + } +} + +seastar::future<> ReplicatedRecoveryBackend::handle_recovery_delete_reply( + Ref<MOSDPGRecoveryDeleteReply> m) +{ + auto& p = m->objects.front(); + hobject_t soid = p.first; + ObjectRecoveryInfo recovery_info; + recovery_info.version = p.second; + pg.get_recovery_handler()->on_peer_recover(m->from, soid, recovery_info); + get_recovering(soid).set_pushed(m->from); + return seastar::now(); +} + +seastar::future<> ReplicatedRecoveryBackend::handle_recovery_op(Ref<MOSDFastDispatchOp> m) +{ + switch (m->get_header().type) { + case MSG_OSD_PG_PULL: + return handle_pull(boost::static_pointer_cast<MOSDPGPull>(m)); + case MSG_OSD_PG_PUSH: + return handle_push(boost::static_pointer_cast<MOSDPGPush>(m)); + case MSG_OSD_PG_PUSH_REPLY: + return handle_push_reply( + boost::static_pointer_cast<MOSDPGPushReply>(m)); + case MSG_OSD_PG_RECOVERY_DELETE: + return handle_recovery_delete( + boost::static_pointer_cast<MOSDPGRecoveryDelete>(m)); + case MSG_OSD_PG_RECOVERY_DELETE_REPLY: + return handle_recovery_delete_reply( + boost::static_pointer_cast<MOSDPGRecoveryDeleteReply>(m)); + default: + // delegate to parent class for handling backend-agnostic recovery ops. + return RecoveryBackend::handle_recovery_op(std::move(m)); + } +} + |