diff options
Diffstat (limited to 'src/mds/ScrubStack.cc')
| -rw-r--r-- | src/mds/ScrubStack.cc | 1191 |
1 files changed, 1191 insertions, 0 deletions
diff --git a/src/mds/ScrubStack.cc b/src/mds/ScrubStack.cc new file mode 100644 index 000000000..6d799343f --- /dev/null +++ b/src/mds/ScrubStack.cc @@ -0,0 +1,1191 @@ +// -*- 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) 2014 Red Hat + * + * 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 "ScrubStack.h" +#include "common/Finisher.h" +#include "mds/MDSRank.h" +#include "mds/MDCache.h" +#include "mds/MDSContinuation.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_mds +#undef dout_prefix +#define dout_prefix _prefix(_dout, mdcache->mds) + +using namespace std; + +static std::ostream& _prefix(std::ostream *_dout, MDSRank *mds) { + return *_dout << "mds." << mds->get_nodeid() << ".scrubstack "; +} + +std::ostream &operator<<(std::ostream &os, const ScrubStack::State &state) { + switch(state) { + case ScrubStack::STATE_RUNNING: + os << "RUNNING"; + break; + case ScrubStack::STATE_IDLE: + os << "IDLE"; + break; + case ScrubStack::STATE_PAUSING: + os << "PAUSING"; + break; + case ScrubStack::STATE_PAUSED: + os << "PAUSED"; + break; + default: + ceph_abort(); + } + + return os; +} + +void ScrubStack::dequeue(MDSCacheObject *obj) +{ + dout(20) << "dequeue " << *obj << " from ScrubStack" << dendl; + ceph_assert(obj->item_scrub.is_on_list()); + obj->put(MDSCacheObject::PIN_SCRUBQUEUE); + obj->item_scrub.remove_myself(); + stack_size--; +} + +int ScrubStack::_enqueue(MDSCacheObject *obj, ScrubHeaderRef& header, bool top) +{ + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + if (CInode *in = dynamic_cast<CInode*>(obj)) { + if (in->scrub_is_in_progress()) { + dout(10) << __func__ << " with {" << *in << "}" << ", already in scrubbing" << dendl; + return -CEPHFS_EBUSY; + } + if(in->state_test(CInode::STATE_PURGING)) { + dout(10) << *obj << " is purging, skip pushing into scrub stack" << dendl; + // treating this as success since purge will make sure this inode goes away + return 0; + } + + dout(10) << __func__ << " with {" << *in << "}" << ", top=" << top << dendl; + in->scrub_initialize(header); + } else if (CDir *dir = dynamic_cast<CDir*>(obj)) { + if (dir->scrub_is_in_progress()) { + dout(10) << __func__ << " with {" << *dir << "}" << ", already in scrubbing" << dendl; + return -CEPHFS_EBUSY; + } + if(dir->get_inode()->state_test(CInode::STATE_PURGING)) { + dout(10) << *obj << " is purging, skip pushing into scrub stack" << dendl; + // treating this as success since purge will make sure this dir inode goes away + return 0; + } + + dout(10) << __func__ << " with {" << *dir << "}" << ", top=" << top << dendl; + // The edge directory must be in memory + dir->auth_pin(this); + dir->scrub_initialize(header); + } else { + ceph_assert(0 == "queue dentry to scrub stack"); + } + + dout(20) << "enqueue " << *obj << " to " << (top ? "top" : "bottom") << " of ScrubStack" << dendl; + if (!obj->item_scrub.is_on_list()) { + obj->get(MDSCacheObject::PIN_SCRUBQUEUE); + stack_size++; + } + if (top) + scrub_stack.push_front(&obj->item_scrub); + else + scrub_stack.push_back(&obj->item_scrub); + return 0; +} + +int ScrubStack::enqueue(CInode *in, ScrubHeaderRef& header, bool top) +{ + // abort in progress + if (clear_stack) + return -CEPHFS_EAGAIN; + + header->set_origin(in->ino()); + auto ret = scrubbing_map.emplace(header->get_tag(), header); + if (!ret.second) { + dout(10) << __func__ << " with {" << *in << "}" + << ", conflicting tag " << header->get_tag() << dendl; + return -CEPHFS_EEXIST; + } + if (header->get_scrub_mdsdir()) { + filepath fp; + mds_rank_t rank; + rank = mdcache->mds->get_nodeid(); + if(rank >= 0 && rank < MAX_MDS) { + fp.set_path("", MDS_INO_MDSDIR(rank)); + } + int r = _enqueue(mdcache->get_inode(fp.get_ino()), header, true); + if (r < 0) { + return r; + } + //to make sure mdsdir is always on the top + top = false; + } + int r = _enqueue(in, header, top); + if (r < 0) + return r; + + clog_scrub_summary(in); + + kick_off_scrubs(); + return 0; +} + +void ScrubStack::add_to_waiting(MDSCacheObject *obj) +{ + scrubs_in_progress++; + obj->item_scrub.remove_myself(); + scrub_waiting.push_back(&obj->item_scrub); +} + +void ScrubStack::remove_from_waiting(MDSCacheObject *obj, bool kick) +{ + scrubs_in_progress--; + if (obj->item_scrub.is_on_list()) { + obj->item_scrub.remove_myself(); + scrub_stack.push_front(&obj->item_scrub); + if (kick) + kick_off_scrubs(); + } +} + +class C_RetryScrub : public MDSInternalContext { +public: + C_RetryScrub(ScrubStack *s, MDSCacheObject *o) : + MDSInternalContext(s->mdcache->mds), stack(s), obj(o) { + stack->add_to_waiting(obj); + } + void finish(int r) override { + stack->remove_from_waiting(obj); + } +private: + ScrubStack *stack; + MDSCacheObject *obj; +}; + +void ScrubStack::kick_off_scrubs() +{ + ceph_assert(ceph_mutex_is_locked(mdcache->mds->mds_lock)); + dout(20) << __func__ << ": state=" << state << dendl; + + if (clear_stack || state == STATE_PAUSING || state == STATE_PAUSED) { + if (scrubs_in_progress == 0) { + dout(10) << __func__ << ": in progress scrub operations finished, " + << stack_size << " in the stack" << dendl; + + State final_state = state; + if (clear_stack) { + abort_pending_scrubs(); + final_state = STATE_IDLE; + } + if (state == STATE_PAUSING) { + final_state = STATE_PAUSED; + } + + set_state(final_state); + complete_control_contexts(0); + } + + return; + } + + dout(20) << __func__ << " entering with " << scrubs_in_progress << " in " + "progress and " << stack_size << " in the stack" << dendl; + elist<MDSCacheObject*>::iterator it = scrub_stack.begin(); + while (g_conf()->mds_max_scrub_ops_in_progress > scrubs_in_progress) { + if (it.end()) { + if (scrubs_in_progress == 0) { + set_state(STATE_IDLE); + } + + return; + } + + assert(state == STATE_RUNNING || state == STATE_IDLE); + set_state(STATE_RUNNING); + + if (CInode *in = dynamic_cast<CInode*>(*it)) { + dout(20) << __func__ << " examining " << *in << dendl; + ++it; + + if (!validate_inode_auth(in)) + continue; + + if (!in->is_dir()) { + // it's a regular file, symlink, or hard link + dequeue(in); // we only touch it this once, so remove from stack + + scrub_file_inode(in); + } else { + bool added_children = false; + bool done = false; // it's done, so pop it off the stack + scrub_dir_inode(in, &added_children, &done); + if (done) { + dout(20) << __func__ << " dir inode, done" << dendl; + dequeue(in); + } + if (added_children) { + // dirfrags were queued at top of stack + it = scrub_stack.begin(); + } + } + } else if (CDir *dir = dynamic_cast<CDir*>(*it)) { + auto next = it; + ++next; + bool done = false; // it's done, so pop it off the stack + scrub_dirfrag(dir, &done); + if (done) { + dout(20) << __func__ << " dirfrag, done" << dendl; + ++it; // child inodes were queued at bottom of stack + dequeue(dir); + } else { + it = next; + } + } else { + ceph_assert(0 == "dentry in scrub stack"); + } + } +} + +bool ScrubStack::validate_inode_auth(CInode *in) +{ + if (in->is_auth()) { + if (!in->can_auth_pin()) { + dout(10) << __func__ << " can't auth pin" << dendl; + in->add_waiter(CInode::WAIT_UNFREEZE, new C_RetryScrub(this, in)); + return false; + } + return true; + } else { + MDSRank *mds = mdcache->mds; + if (in->is_ambiguous_auth()) { + dout(10) << __func__ << " ambiguous auth" << dendl; + in->add_waiter(MDSCacheObject::WAIT_SINGLEAUTH, new C_RetryScrub(this, in)); + } else if (mds->is_cluster_degraded()) { + dout(20) << __func__ << " cluster degraded" << dendl; + mds->wait_for_cluster_recovered(new C_RetryScrub(this, in)); + } else { + ScrubHeaderRef header = in->get_scrub_header(); + ceph_assert(header); + + auto ret = remote_scrubs.emplace(std::piecewise_construct, + std::forward_as_tuple(in), + std::forward_as_tuple()); + ceph_assert(ret.second); // FIXME: parallel scrubs? + auto &scrub_r = ret.first->second; + scrub_r.tag = header->get_tag(); + + mds_rank_t auth = in->authority().first; + dout(10) << __func__ << " forward to mds." << auth << dendl; + auto r = make_message<MMDSScrub>(MMDSScrub::OP_QUEUEINO, in->ino(), + std::move(in->scrub_queued_frags()), + header->get_tag(), header->get_origin(), + header->is_internal_tag(), header->get_force(), + header->get_recursive(), header->get_repair()); + mdcache->mds->send_message_mds(r, auth); + + scrub_r.gather_set.insert(auth); + // wait for ACK + add_to_waiting(in); + } + return false; + } +} + +void ScrubStack::scrub_dir_inode(CInode *in, bool *added_children, bool *done) +{ + dout(10) << __func__ << " " << *in << dendl; + ceph_assert(in->is_auth()); + MDSRank *mds = mdcache->mds; + + ScrubHeaderRef header = in->get_scrub_header(); + ceph_assert(header); + + MDSGatherBuilder gather(g_ceph_context); + + auto &queued = in->scrub_queued_frags(); + std::map<mds_rank_t, fragset_t> scrub_remote; + + frag_vec_t frags; + in->dirfragtree.get_leaves(frags); + dout(20) << __func__ << "recursive mode, frags " << frags << dendl; + for (auto &fg : frags) { + if (queued.contains(fg)) + continue; + CDir *dir = in->get_or_open_dirfrag(mdcache, fg); + if (!dir->is_auth()) { + if (dir->is_ambiguous_auth()) { + dout(20) << __func__ << " ambiguous auth " << *dir << dendl; + dir->add_waiter(MDSCacheObject::WAIT_SINGLEAUTH, gather.new_sub()); + } else if (mds->is_cluster_degraded()) { + dout(20) << __func__ << " cluster degraded" << dendl; + mds->wait_for_cluster_recovered(gather.new_sub()); + } else { + mds_rank_t auth = dir->authority().first; + scrub_remote[auth].insert_raw(fg); + } + } else if (!dir->can_auth_pin()) { + dout(20) << __func__ << " freezing/frozen " << *dir << dendl; + dir->add_waiter(CDir::WAIT_UNFREEZE, gather.new_sub()); + } else if (dir->get_version() == 0) { + dout(20) << __func__ << " barebones " << *dir << dendl; + dir->fetch_keys({}, gather.new_sub()); + } else { + _enqueue(dir, header, true); + queued.insert_raw(dir->get_frag()); + *added_children = true; + } + } + + queued.simplify(); + + if (gather.has_subs()) { + gather.set_finisher(new C_RetryScrub(this, in)); + gather.activate(); + return; + } + + if (!scrub_remote.empty()) { + auto ret = remote_scrubs.emplace(std::piecewise_construct, + std::forward_as_tuple(in), + std::forward_as_tuple()); + ceph_assert(ret.second); // FIXME: parallel scrubs? + auto &scrub_r = ret.first->second; + scrub_r.tag = header->get_tag(); + + for (auto& p : scrub_remote) { + p.second.simplify(); + dout(20) << __func__ << " forward " << p.second << " to mds." << p.first << dendl; + auto r = make_message<MMDSScrub>(MMDSScrub::OP_QUEUEDIR, in->ino(), + std::move(p.second), header->get_tag(), + header->get_origin(), header->is_internal_tag(), + header->get_force(), header->get_recursive(), + header->get_repair()); + mds->send_message_mds(r, p.first); + scrub_r.gather_set.insert(p.first); + } + // wait for ACKs + add_to_waiting(in); + return; + } + + scrub_dir_inode_final(in); + + *done = true; + dout(10) << __func__ << " done" << dendl; +} + +class C_InodeValidated : public MDSInternalContext +{ +public: + ScrubStack *stack; + CInode::validated_data result; + CInode *target; + + C_InodeValidated(MDSRank *mds, ScrubStack *stack_, CInode *target_) + : MDSInternalContext(mds), stack(stack_), target(target_) + { + stack->scrubs_in_progress++; + } + void finish(int r) override { + stack->_validate_inode_done(target, r, result); + stack->scrubs_in_progress--; + stack->kick_off_scrubs(); + } +}; + +void ScrubStack::scrub_dir_inode_final(CInode *in) +{ + dout(20) << __func__ << " " << *in << dendl; + + C_InodeValidated *fin = new C_InodeValidated(mdcache->mds, this, in); + in->validate_disk_state(&fin->result, fin); + return; +} + +void ScrubStack::scrub_dirfrag(CDir *dir, bool *done) +{ + ceph_assert(dir != NULL); + + dout(10) << __func__ << " " << *dir << dendl; + + if (!dir->is_complete()) { + dir->fetch(new C_RetryScrub(this, dir), true); // already auth pinned + dout(10) << __func__ << " incomplete, fetching" << dendl; + return; + } + + ScrubHeaderRef header = dir->get_scrub_header(); + version_t last_scrub = dir->scrub_info()->last_recursive.version; + if (header->get_recursive()) { + auto next_seq = mdcache->get_global_snaprealm()->get_newest_seq()+1; + for (auto it = dir->begin(); it != dir->end(); /* nop */) { + auto [dnk, dn] = *it; + ++it; /* trim (in the future) may remove dentry */ + + if (dn->scrub(next_seq)) { + std::string path; + dir->get_inode()->make_path_string(path, true); + clog->warn() << "Scrub error on dentry " << *dn + << " see " << g_conf()->name + << " log and `damage ls` output for details"; + } + + if (dnk.snapid != CEPH_NOSNAP) { + continue; + } + + CDentry::linkage_t *dnl = dn->get_linkage(); + if (dn->get_version() <= last_scrub && + dnl->get_remote_d_type() != DT_DIR && + !header->get_force()) { + dout(15) << __func__ << " skip dentry " << dnk + << ", no change since last scrub" << dendl; + continue; + } + if (dnl->is_primary()) { + _enqueue(dnl->get_inode(), header, false); + } else if (dnl->is_remote()) { + // TODO: check remote linkage + } + } + } + + if (!dir->scrub_local()) { + std::string path; + dir->get_inode()->make_path_string(path, true); + clog->warn() << "Scrub error on dir " << dir->ino() + << " (" << path << ") see " << g_conf()->name + << " log and `damage ls` output for details"; + } + + dir->scrub_finished(); + dir->auth_unpin(this); + + *done = true; + dout(10) << __func__ << " done" << dendl; +} + +void ScrubStack::scrub_file_inode(CInode *in) +{ + C_InodeValidated *fin = new C_InodeValidated(mdcache->mds, this, in); + // At this stage the DN is already past scrub_initialize, so + // it's in the cache, it has PIN_SCRUBQUEUE and it is authpinned + in->validate_disk_state(&fin->result, fin); +} + +void ScrubStack::_validate_inode_done(CInode *in, int r, + const CInode::validated_data &result) +{ + LogChannelRef clog = mdcache->mds->clog; + const ScrubHeaderRefConst header = in->scrub_info()->header; + + std::string path; + if (!result.passed_validation) { + // Build path string for use in messages + in->make_path_string(path, true); + } + + if (result.backtrace.checked && !result.backtrace.passed && + !result.backtrace.repaired) + { + // Record backtrace fails as remote linkage damage, as + // we may not be able to resolve hard links to this inode + mdcache->mds->damage_table.notify_remote_damaged(in->ino(), path); + } else if (result.inode.checked && !result.inode.passed && + !result.inode.repaired) { + // Record damaged inode structures as damaged dentries as + // that is where they are stored + auto parent = in->get_projected_parent_dn(); + if (parent) { + auto dir = parent->get_dir(); + mdcache->mds->damage_table.notify_dentry( + dir->inode->ino(), dir->frag, parent->last, parent->get_name(), path); + } + } + + // Inform the cluster log if we found an error + if (!result.passed_validation) { + if (result.all_damage_repaired()) { + clog->info() << "Scrub repaired inode " << in->ino() + << " (" << path << ")"; + } else { + clog->warn() << "Scrub error on inode " << in->ino() + << " (" << path << ") see " << g_conf()->name + << " log and `damage ls` output for details"; + } + + // Put the verbose JSON output into the MDS log for later inspection + JSONFormatter f; + result.dump(&f); + CachedStackStringStream css; + f.flush(*css); + derr << __func__ << " scrub error on inode " << *in << ": " << css->strv() + << dendl; + } else { + dout(10) << __func__ << " scrub passed on inode " << *in << dendl; + } + + in->scrub_finished(); +} + +void ScrubStack::complete_control_contexts(int r) { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + + for (auto &ctx : control_ctxs) { + ctx->complete(r); + } + control_ctxs.clear(); +} + +void ScrubStack::set_state(State next_state) { + if (state != next_state) { + dout(20) << __func__ << ", from state=" << state << ", to state=" + << next_state << dendl; + state = next_state; + clog_scrub_summary(); + } +} + +bool ScrubStack::scrub_in_transition_state() { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + dout(20) << __func__ << ": state=" << state << dendl; + + // STATE_RUNNING is considered as a transition state so as to + // "delay" the scrub control operation. + if (state == STATE_RUNNING || state == STATE_PAUSING) { + return true; + } + + return false; +} + +std::string_view ScrubStack::scrub_summary() { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + + bool have_more = false; + CachedStackStringStream cs; + + if (state == STATE_IDLE) { + if (scrubbing_map.empty()) + return "idle"; + *cs << "idle+waiting"; + } + + if (state == STATE_RUNNING) { + if (clear_stack) { + *cs << "aborting"; + } else { + *cs << "active"; + } + } else { + if (state == STATE_PAUSING) { + have_more = true; + *cs << "pausing"; + } else if (state == STATE_PAUSED) { + have_more = true; + *cs << "paused"; + } + + if (clear_stack) { + if (have_more) { + *cs << "+"; + } + *cs << "aborting"; + } + } + + if (!scrubbing_map.empty()) { + *cs << " paths ["; + bool first = true; + for (auto &p : scrubbing_map) { + if (!first) + *cs << ","; + auto& header = p.second; + if (CInode *in = mdcache->get_inode(header->get_origin())) + *cs << scrub_inode_path(in); + else + *cs << "#" << header->get_origin(); + first = false; + } + *cs << "]"; + } + + return cs->strv(); +} + +void ScrubStack::scrub_status(Formatter *f) { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + + f->open_object_section("result"); + + CachedStackStringStream css; + bool have_more = false; + + if (state == STATE_IDLE) { + if (scrubbing_map.empty()) + *css << "no active scrubs running"; + else + *css << state << " (waiting for more scrubs)"; + } else if (state == STATE_RUNNING) { + if (clear_stack) { + *css << "ABORTING"; + } else { + *css << "scrub active"; + } + *css << " (" << stack_size << " inodes in the stack)"; + } else { + if (state == STATE_PAUSING || state == STATE_PAUSED) { + have_more = true; + *css << state; + } + if (clear_stack) { + if (have_more) { + *css << "+"; + } + *css << "ABORTING"; + } + + *css << " (" << stack_size << " inodes in the stack)"; + } + f->dump_string("status", css->strv()); + + f->open_object_section("scrubs"); + + for (auto& p : scrubbing_map) { + have_more = false; + auto& header = p.second; + + std::string tag(header->get_tag()); + f->open_object_section(tag.c_str()); // scrub id + + if (CInode *in = mdcache->get_inode(header->get_origin())) + f->dump_string("path", scrub_inode_path(in)); + else + f->dump_stream("path") << "#" << header->get_origin(); + + f->dump_string("tag", header->get_tag()); + + CachedStackStringStream optcss; + if (header->get_recursive()) { + *optcss << "recursive"; + have_more = true; + } + if (header->get_repair()) { + if (have_more) { + *optcss << ","; + } + *optcss << "repair"; + have_more = true; + } + if (header->get_force()) { + if (have_more) { + *optcss << ","; + } + *optcss << "force"; + } + if (header->get_scrub_mdsdir()) { + if (have_more) { + *optcss << ","; + } + *optcss << "scrub_mdsdir"; + } + + f->dump_string("options", optcss->strv()); + f->close_section(); // scrub id + } + f->close_section(); // scrubs + f->close_section(); // result +} + +void ScrubStack::abort_pending_scrubs() { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + ceph_assert(clear_stack); + + auto abort_one = [this](MDSCacheObject *obj) { + if (CInode *in = dynamic_cast<CInode*>(obj)) { + in->scrub_aborted(); + } else if (CDir *dir = dynamic_cast<CDir*>(obj)) { + dir->scrub_aborted(); + dir->auth_unpin(this); + } else { + ceph_abort(0 == "dentry in scrub stack"); + } + }; + for (auto it = scrub_stack.begin(); !it.end(); ++it) + abort_one(*it); + for (auto it = scrub_waiting.begin(); !it.end(); ++it) + abort_one(*it); + + stack_size = 0; + scrub_stack.clear(); + scrub_waiting.clear(); + + for (auto& p : remote_scrubs) + remove_from_waiting(p.first, false); + remote_scrubs.clear(); + + clear_stack = false; +} + +void ScrubStack::send_state_message(int op) { + MDSRank *mds = mdcache->mds; + set<mds_rank_t> up_mds; + mds->get_mds_map()->get_up_mds_set(up_mds); + for (auto& r : up_mds) { + if (r == 0) + continue; + auto m = make_message<MMDSScrub>(op); + mds->send_message_mds(m, r); + } +} + +void ScrubStack::scrub_abort(Context *on_finish) { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + + dout(10) << __func__ << ": aborting with " << scrubs_in_progress + << " scrubs in progress and " << stack_size << " in the" + << " stack" << dendl; + + if (mdcache->mds->get_nodeid() == 0) { + scrub_epoch_last_abort = scrub_epoch; + scrub_any_peer_aborting = true; + send_state_message(MMDSScrub::OP_ABORT); + } + + clear_stack = true; + if (scrub_in_transition_state()) { + if (on_finish) + control_ctxs.push_back(on_finish); + return; + } + + abort_pending_scrubs(); + if (state != STATE_PAUSED) + set_state(STATE_IDLE); + + if (on_finish) + on_finish->complete(0); +} + +void ScrubStack::scrub_pause(Context *on_finish) { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + + dout(10) << __func__ << ": pausing with " << scrubs_in_progress + << " scrubs in progress and " << stack_size << " in the" + << " stack" << dendl; + + if (mdcache->mds->get_nodeid() == 0) + send_state_message(MMDSScrub::OP_PAUSE); + + // abort is in progress + if (clear_stack) { + if (on_finish) + on_finish->complete(-CEPHFS_EINVAL); + return; + } + + bool done = scrub_in_transition_state(); + if (done) { + set_state(STATE_PAUSING); + if (on_finish) + control_ctxs.push_back(on_finish); + return; + } + + set_state(STATE_PAUSED); + if (on_finish) + on_finish->complete(0); +} + +bool ScrubStack::scrub_resume() { + ceph_assert(ceph_mutex_is_locked_by_me(mdcache->mds->mds_lock)); + dout(20) << __func__ << ": state=" << state << dendl; + + if (mdcache->mds->get_nodeid() == 0) + send_state_message(MMDSScrub::OP_RESUME); + + int r = 0; + + if (clear_stack) { + r = -CEPHFS_EINVAL; + } else if (state == STATE_PAUSING) { + set_state(STATE_RUNNING); + complete_control_contexts(-CEPHFS_ECANCELED); + } else if (state == STATE_PAUSED) { + set_state(STATE_RUNNING); + kick_off_scrubs(); + } + + return r; +} + +// send current scrub summary to cluster log +void ScrubStack::clog_scrub_summary(CInode *in) { + if (in) { + std::string what; + if (clear_stack) { + what = "aborted"; + } else if (in->scrub_is_in_progress()) { + what = "queued"; + } else { + what = "completed"; + } + clog->info() << "scrub " << what << " for path: " << scrub_inode_path(in); + } + + clog->info() << "scrub summary: " << scrub_summary(); +} + +void ScrubStack::dispatch(const cref_t<Message> &m) +{ + switch (m->get_type()) { + case MSG_MDS_SCRUB: + handle_scrub(ref_cast<MMDSScrub>(m)); + break; + + case MSG_MDS_SCRUB_STATS: + handle_scrub_stats(ref_cast<MMDSScrubStats>(m)); + break; + + default: + derr << " scrub stack unknown message " << m->get_type() << dendl_impl; + ceph_abort_msg("scrub stack unknown message"); + } +} + +bool ScrubStack::remove_inode_if_stacked(CInode *in) { + MDSCacheObject *obj = dynamic_cast<MDSCacheObject*>(in); + if(obj->item_scrub.is_on_list()) { + dout(20) << "removing inode " << *in << " from scrub_stack" << dendl; + obj->put(MDSCacheObject::PIN_SCRUBQUEUE); + obj->item_scrub.remove_myself(); + stack_size--; + return true; + } + return false; +} + +void ScrubStack::handle_scrub(const cref_t<MMDSScrub> &m) +{ + + mds_rank_t from = mds_rank_t(m->get_source().num()); + dout(10) << __func__ << " " << *m << " from mds." << from << dendl; + + switch (m->get_op()) { + case MMDSScrub::OP_QUEUEDIR: + { + CInode *diri = mdcache->get_inode(m->get_ino()); + ceph_assert(diri); + + std::vector<CDir*> dfs; + MDSGatherBuilder gather(g_ceph_context); + for (const auto& fg : m->get_frags()) { + CDir *dir = diri->get_dirfrag(fg); + if (!dir) { + dout(10) << __func__ << " no frag " << fg << dendl; + continue; + } + if (!dir->is_auth()) { + dout(10) << __func__ << " not auth " << *dir << dendl; + continue; + } + if (!dir->can_auth_pin()) { + dout(10) << __func__ << " can't auth pin " << *dir << dendl; + dir->add_waiter(CDir::WAIT_UNFREEZE, gather.new_sub()); + continue; + } + dfs.push_back(dir); + } + + if (gather.has_subs()) { + gather.set_finisher(new C_MDS_RetryMessage(mdcache->mds, m)); + gather.activate(); + return; + } + + fragset_t queued; + if (!dfs.empty()) { + ScrubHeaderRef header; + if (auto it = scrubbing_map.find(m->get_tag()); it != scrubbing_map.end()) { + header = it->second; + } else { + header = std::make_shared<ScrubHeader>(m->get_tag(), m->is_internal_tag(), + m->is_force(), m->is_recursive(), + m->is_repair()); + header->set_origin(m->get_origin()); + scrubbing_map.emplace(header->get_tag(), header); + } + for (auto dir : dfs) { + queued.insert_raw(dir->get_frag()); + _enqueue(dir, header, true); + } + queued.simplify(); + kick_off_scrubs(); + } + + auto r = make_message<MMDSScrub>(MMDSScrub::OP_QUEUEDIR_ACK, m->get_ino(), + std::move(queued), m->get_tag()); + mdcache->mds->send_message_mds(r, from); + } + break; + case MMDSScrub::OP_QUEUEDIR_ACK: + { + CInode *diri = mdcache->get_inode(m->get_ino()); + ceph_assert(diri); + auto it = remote_scrubs.find(diri); + if (it != remote_scrubs.end() && + m->get_tag() == it->second.tag) { + if (it->second.gather_set.erase(from)) { + auto &queued = diri->scrub_queued_frags(); + for (auto &fg : m->get_frags()) + queued.insert_raw(fg); + queued.simplify(); + + if (it->second.gather_set.empty()) { + remote_scrubs.erase(it); + + const auto& header = diri->get_scrub_header(); + header->set_epoch_last_forwarded(scrub_epoch); + remove_from_waiting(diri); + } + } + } + } + break; + case MMDSScrub::OP_QUEUEINO: + { + CInode *in = mdcache->get_inode(m->get_ino()); + ceph_assert(in); + + ScrubHeaderRef header; + if (auto it = scrubbing_map.find(m->get_tag()); it != scrubbing_map.end()) { + header = it->second; + } else { + header = std::make_shared<ScrubHeader>(m->get_tag(), m->is_internal_tag(), + m->is_force(), m->is_recursive(), + m->is_repair()); + header->set_origin(m->get_origin()); + scrubbing_map.emplace(header->get_tag(), header); + } + + _enqueue(in, header, true); + in->scrub_queued_frags() = m->get_frags(); + kick_off_scrubs(); + + fragset_t queued; + auto r = make_message<MMDSScrub>(MMDSScrub::OP_QUEUEINO_ACK, m->get_ino(), + std::move(queued), m->get_tag()); + mdcache->mds->send_message_mds(r, from); + } + break; + case MMDSScrub::OP_QUEUEINO_ACK: + { + CInode *in = mdcache->get_inode(m->get_ino()); + ceph_assert(in); + auto it = remote_scrubs.find(in); + if (it != remote_scrubs.end() && + m->get_tag() == it->second.tag && + it->second.gather_set.erase(from)) { + ceph_assert(it->second.gather_set.empty()); + remote_scrubs.erase(it); + + remove_from_waiting(in, false); + dequeue(in); + + const auto& header = in->get_scrub_header(); + header->set_epoch_last_forwarded(scrub_epoch); + in->scrub_finished(); + + kick_off_scrubs(); + } + } + break; + case MMDSScrub::OP_ABORT: + scrub_abort(nullptr); + break; + case MMDSScrub::OP_PAUSE: + scrub_pause(nullptr); + break; + case MMDSScrub::OP_RESUME: + scrub_resume(); + break; + default: + derr << " scrub stack unknown scrub operation " << m->get_op() << dendl_impl; + ceph_abort_msg("scrub stack unknown scrub operation"); + } +} + +void ScrubStack::handle_scrub_stats(const cref_t<MMDSScrubStats> &m) +{ + mds_rank_t from = mds_rank_t(m->get_source().num()); + dout(7) << __func__ << " " << *m << " from mds." << from << dendl; + + if (from == 0) { + if (scrub_epoch != m->get_epoch() - 1) { + scrub_epoch = m->get_epoch() - 1; + for (auto& p : scrubbing_map) { + if (p.second->get_epoch_last_forwarded()) + p.second->set_epoch_last_forwarded(scrub_epoch); + } + } + bool any_finished = false; + bool any_repaired = false; + std::set<std::string> scrubbing_tags; + for (auto it = scrubbing_map.begin(); it != scrubbing_map.end(); ) { + auto& header = it->second; + if (header->get_num_pending() || + header->get_epoch_last_forwarded() >= scrub_epoch) { + scrubbing_tags.insert(it->first); + ++it; + } else if (m->is_finished(it->first)) { + any_finished = true; + if (header->get_repaired()) + any_repaired = true; + scrubbing_map.erase(it++); + } else { + ++it; + } + } + + scrub_epoch = m->get_epoch(); + + auto ack = make_message<MMDSScrubStats>(scrub_epoch, + std::move(scrubbing_tags), clear_stack); + mdcache->mds->send_message_mds(ack, 0); + + if (any_finished) + clog_scrub_summary(); + if (any_repaired) + mdcache->mds->mdlog->trim_all(); + } else { + if (scrub_epoch == m->get_epoch() && + (size_t)from < mds_scrub_stats.size()) { + auto& stat = mds_scrub_stats[from]; + stat.epoch_acked = m->get_epoch(); + stat.scrubbing_tags = m->get_scrubbing_tags(); + stat.aborting = m->is_aborting(); + } + } +} + +void ScrubStack::advance_scrub_status() +{ + if (!scrub_any_peer_aborting && scrubbing_map.empty()) + return; + + MDSRank *mds = mdcache->mds; + + set<mds_rank_t> up_mds; + mds->get_mds_map()->get_up_mds_set(up_mds); + auto up_max = *up_mds.rbegin(); + + bool update_scrubbing = false; + std::set<std::string> scrubbing_tags; + + if (up_max == 0) { + update_scrubbing = true; + scrub_any_peer_aborting = false; + } else if (mds_scrub_stats.size() > (size_t)(up_max)) { + bool any_aborting = false; + bool fully_acked = true; + for (const auto& stat : mds_scrub_stats) { + if (stat.aborting || stat.epoch_acked <= scrub_epoch_last_abort) + any_aborting = true; + if (stat.epoch_acked != scrub_epoch) { + fully_acked = false; + continue; + } + scrubbing_tags.insert(stat.scrubbing_tags.begin(), + stat.scrubbing_tags.end()); + } + if (!any_aborting) + scrub_any_peer_aborting = false; + if (fully_acked) { + // handle_scrub_stats() reports scrub is still in-progress if it has + // forwarded any object to other mds since previous epoch. Let's assume, + // at time 'A', we got scrub stats from all mds for previous epoch. If + // a scrub is not reported by any mds, we know there is no forward of + // the scrub since time 'A'. So we can consider the scrub is finished. + if (scrub_epoch_fully_acked + 1 == scrub_epoch) + update_scrubbing = true; + scrub_epoch_fully_acked = scrub_epoch; + } + } + + if (mds_scrub_stats.size() != (size_t)up_max + 1) + mds_scrub_stats.resize((size_t)up_max + 1); + mds_scrub_stats.at(0).epoch_acked = scrub_epoch + 1; + + bool any_finished = false; + bool any_repaired = false; + + for (auto it = scrubbing_map.begin(); it != scrubbing_map.end(); ) { + auto& header = it->second; + if (header->get_num_pending() || + header->get_epoch_last_forwarded() >= scrub_epoch) { + if (update_scrubbing && up_max != 0) + scrubbing_tags.insert(it->first); + ++it; + } else if (update_scrubbing && !scrubbing_tags.count(it->first)) { + // no longer being scrubbed globally + any_finished = true; + if (header->get_repaired()) + any_repaired = true; + scrubbing_map.erase(it++); + } else { + ++it; + } + } + + ++scrub_epoch; + + for (auto& r : up_mds) { + if (r == 0) + continue; + auto m = update_scrubbing ? + make_message<MMDSScrubStats>(scrub_epoch, scrubbing_tags) : + make_message<MMDSScrubStats>(scrub_epoch); + mds->send_message_mds(m, r); + } + + if (any_finished) + clog_scrub_summary(); + if (any_repaired) + mdcache->mds->mdlog->trim_all(); +} + +void ScrubStack::handle_mds_failure(mds_rank_t mds) +{ + if (mds == 0) { + scrub_abort(nullptr); + return; + } + + bool kick = false; + for (auto it = remote_scrubs.begin(); it != remote_scrubs.end(); ) { + if (it->second.gather_set.erase(mds) && + it->second.gather_set.empty()) { + CInode *in = it->first; + remote_scrubs.erase(it++); + remove_from_waiting(in, false); + kick = true; + } else { + ++it; + } + } + if (kick) + kick_off_scrubs(); +} |