diff options
Diffstat (limited to 'src/mds/MDSDaemon.cc')
| -rw-r--r-- | src/mds/MDSDaemon.cc | 1154 |
1 files changed, 1154 insertions, 0 deletions
diff --git a/src/mds/MDSDaemon.cc b/src/mds/MDSDaemon.cc new file mode 100644 index 000000000..d45acce06 --- /dev/null +++ b/src/mds/MDSDaemon.cc @@ -0,0 +1,1154 @@ +// -*- 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) 2004-2006 Sage Weil <sage@newdream.net> + * + * 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 <unistd.h> + +#include "include/compat.h" +#include "include/types.h" +#include "include/str_list.h" + +#include "common/Clock.h" +#include "common/HeartbeatMap.h" +#include "common/Timer.h" +#include "common/ceph_argparse.h" +#include "common/config.h" +#include "common/entity_name.h" +#include "common/errno.h" +#include "common/perf_counters.h" +#include "common/signal.h" +#include "common/version.h" + +#include "global/signal_handler.h" + +#include "msg/Messenger.h" +#include "mon/MonClient.h" + +#include "osdc/Objecter.h" + +#include "MDSMap.h" + +#include "MDSDaemon.h" +#include "Server.h" +#include "Locker.h" + +#include "SnapServer.h" +#include "SnapClient.h" + +#include "events/ESession.h" +#include "events/ESubtreeMap.h" + +#include "auth/AuthAuthorizeHandler.h" +#include "auth/RotatingKeyRing.h" +#include "auth/KeyRing.h" + +#include "perfglue/cpu_profiler.h" +#include "perfglue/heap_profiler.h" + +#define dout_context g_ceph_context +#define dout_subsys ceph_subsys_mds +#undef dout_prefix +#define dout_prefix *_dout << "mds." << name << ' ' + +using std::string; +using std::vector; +using TOPNSPC::common::cmd_getval; + +// cons/des +MDSDaemon::MDSDaemon(std::string_view n, Messenger *m, MonClient *mc, + boost::asio::io_context& ioctx) : + Dispatcher(m->cct), + timer(m->cct, mds_lock), + gss_ktfile_client(m->cct->_conf.get_val<std::string>("gss_ktab_client_file")), + beacon(m->cct, mc, n), + name(n), + messenger(m), + monc(mc), + ioctx(ioctx), + mgrc(m->cct, m, &mc->monmap), + log_client(m->cct, messenger, &mc->monmap, LogClient::NO_FLAGS), + starttime(mono_clock::now()) +{ + orig_argc = 0; + orig_argv = NULL; + + clog = log_client.create_channel(); + if (!gss_ktfile_client.empty()) { + // Assert we can export environment variable + /* + The default client keytab is used, if it is present and readable, + to automatically obtain initial credentials for GSSAPI client + applications. The principal name of the first entry in the client + keytab is used by default when obtaining initial credentials. + 1. The KRB5_CLIENT_KTNAME environment variable. + 2. The default_client_keytab_name profile variable in [libdefaults]. + 3. The hardcoded default, DEFCKTNAME. + */ + const int32_t set_result(setenv("KRB5_CLIENT_KTNAME", + gss_ktfile_client.c_str(), 1)); + ceph_assert(set_result == 0); + } + + mdsmap.reset(new MDSMap); +} + +MDSDaemon::~MDSDaemon() { + std::lock_guard lock(mds_lock); + + delete mds_rank; + mds_rank = NULL; +} + +class MDSSocketHook : public AdminSocketHook { + MDSDaemon *mds; +public: + explicit MDSSocketHook(MDSDaemon *m) : mds(m) {} + int call( + std::string_view command, + const cmdmap_t& cmdmap, + const bufferlist&, + Formatter *f, + std::ostream& errss, + ceph::buffer::list& out) override { + ceph_abort("should go to call_async"); + } + void call_async( + std::string_view command, + const cmdmap_t& cmdmap, + Formatter *f, + const bufferlist& inbl, + std::function<void(int,const std::string&,bufferlist&)> on_finish) override { + mds->asok_command(command, cmdmap, f, inbl, on_finish); + } +}; + +void MDSDaemon::asok_command( + std::string_view command, + const cmdmap_t& cmdmap, + Formatter *f, + const bufferlist& inbl, + std::function<void(int,const std::string&,bufferlist&)> on_finish) +{ + dout(1) << "asok_command: " << command << " " << cmdmap + << " (starting...)" << dendl; + + int r = -CEPHFS_ENOSYS; + bufferlist outbl; + CachedStackStringStream css; + auto& ss = *css; + if (command == "status") { + dump_status(f); + r = 0; + } else if (command == "exit") { + outbl.append("Exiting...\n"); + r = 0; + std::thread t([this](){ + // Wait a little to improve chances of caller getting + // our response before seeing us disappear from mdsmap + sleep(1); + std::lock_guard l(mds_lock); + derr << "Exiting due to admin socket command" << dendl; + suicide(); + }); + t.detach(); + } else if (command == "respawn") { + outbl.append("Respawning...\n"); + r = 0; + std::thread t([this](){ + // Wait a little to improve chances of caller getting + // our response before seeing us disappear from mdsmap + sleep(1); + std::lock_guard l(mds_lock); + respawn(); + }); + t.detach(); + } else if (command == "heap") { + if (!ceph_using_tcmalloc()) { + ss << "not using tcmalloc"; + r = -CEPHFS_EOPNOTSUPP; + } else { + string heapcmd; + cmd_getval(cmdmap, "heapcmd", heapcmd); + vector<string> heapcmd_vec; + get_str_vec(heapcmd, heapcmd_vec); + string value; + if (cmd_getval(cmdmap, "value", value)) { + heapcmd_vec.push_back(value); + } + std::stringstream outss; + ceph_heap_profiler_handle_command(heapcmd_vec, outss); + outbl.append(outss); + r = 0; + } + } else if (command == "cpu_profiler") { + string arg; + cmd_getval(cmdmap, "arg", arg); + vector<string> argvec; + get_str_vec(arg, argvec); + cpu_profiler_handle_command(argvec, ss); + r = 0; + } else { + if (mds_rank == NULL) { + dout(1) << "Can't run that command on an inactive MDS!" << dendl; + f->dump_string("error", "mds_not_active"); + } else { + try { + mds_rank->handle_asok_command(command, cmdmap, f, inbl, on_finish); + return; + } catch (const TOPNSPC::common::bad_cmd_get& e) { + ss << e.what(); + r = -CEPHFS_EINVAL; + } + } + } + on_finish(r, ss.str(), outbl); +} + +void MDSDaemon::dump_status(Formatter *f) +{ + f->open_object_section("status"); + f->dump_stream("cluster_fsid") << monc->get_fsid(); + if (mds_rank) { + f->dump_int("whoami", mds_rank->get_nodeid()); + } else { + f->dump_int("whoami", MDS_RANK_NONE); + } + + f->dump_int("id", monc->get_global_id()); + f->dump_string("want_state", ceph_mds_state_name(beacon.get_want_state())); + f->dump_string("state", ceph_mds_state_name(mdsmap->get_state_gid(mds_gid_t( + monc->get_global_id())))); + if (mds_rank) { + std::lock_guard l(mds_lock); + mds_rank->dump_status(f); + } + + f->dump_unsigned("mdsmap_epoch", mdsmap->get_epoch()); + if (mds_rank) { + f->dump_unsigned("osdmap_epoch", mds_rank->get_osd_epoch()); + f->dump_unsigned("osdmap_epoch_barrier", mds_rank->get_osd_epoch_barrier()); + } else { + f->dump_unsigned("osdmap_epoch", 0); + f->dump_unsigned("osdmap_epoch_barrier", 0); + } + + f->dump_float("uptime", get_uptime().count()); + + f->close_section(); // status +} + +void MDSDaemon::set_up_admin_socket() +{ + int r; + AdminSocket *admin_socket = g_ceph_context->get_admin_socket(); + ceph_assert(asok_hook == nullptr); + asok_hook = new MDSSocketHook(this); + r = admin_socket->register_command("status", asok_hook, + "high-level status of MDS"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_ops_in_flight", asok_hook, + "show the ops currently in flight"); + ceph_assert(r == 0); + r = admin_socket->register_command("ops " + "name=flags,type=CephChoices,strings=locks,n=N,req=false ", + asok_hook, + "show the ops currently in flight"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_blocked_ops", + asok_hook, + "show the blocked ops currently in flight"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_blocked_ops_count", + asok_hook, + "show the count of blocked ops currently in flight"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_historic_ops", + asok_hook, + "show recent ops"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_historic_ops_by_duration", + asok_hook, + "show recent ops, sorted by op duration"); + ceph_assert(r == 0); + r = admin_socket->register_command("scrub_path name=path,type=CephString " + "name=scrubops,type=CephChoices," + "strings=force|recursive|repair,n=N,req=false " + "name=tag,type=CephString,req=false", + asok_hook, + "scrub an inode and output results"); + ceph_assert(r == 0); + r = admin_socket->register_command("scrub start " + "name=path,type=CephString " + "name=scrubops,type=CephChoices,strings=force|recursive|repair|scrub_mdsdir,n=N,req=false " + "name=tag,type=CephString,req=false", + asok_hook, + "scrub and inode and output results"); + ceph_assert(r == 0); + r = admin_socket->register_command("scrub abort", + asok_hook, + "Abort in progress scrub operations(s)"); + ceph_assert(r == 0); + r = admin_socket->register_command("scrub pause", + asok_hook, + "Pause in progress scrub operations(s)"); + ceph_assert(r == 0); + r = admin_socket->register_command("scrub resume", + asok_hook, + "Resume paused scrub operations(s)"); + ceph_assert(r == 0); + r = admin_socket->register_command("scrub status", + asok_hook, + "Status of scrub operations(s)"); + ceph_assert(r == 0); + r = admin_socket->register_command("tag path name=path,type=CephString" + " name=tag,type=CephString", + asok_hook, + "Apply scrub tag recursively"); + ceph_assert(r == 0); + r = admin_socket->register_command("flush_path name=path,type=CephString", + asok_hook, + "flush an inode (and its dirfrags)"); + ceph_assert(r == 0); + r = admin_socket->register_command("export dir " + "name=path,type=CephString " + "name=rank,type=CephInt", + asok_hook, + "migrate a subtree to named MDS"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump cache " + "name=path,type=CephString,req=false " + "name=timeout,type=CephInt,range=0,req=false", + asok_hook, + "dump metadata cache (optionally to a file)"); + ceph_assert(r == 0); + r = admin_socket->register_command("cache drop " + "name=timeout,type=CephInt,range=0,req=false", + asok_hook, + "trim cache and optionally request client to release all caps and flush the journal"); + ceph_assert(r == 0); + r = admin_socket->register_command("cache status", + asok_hook, + "show cache status"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump tree " + "name=root,type=CephString,req=true " + "name=depth,type=CephInt,req=false ", + asok_hook, + "dump metadata cache for subtree"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump loads " + "name=depth,type=CephInt,range=0,req=false", + asok_hook, + "dump metadata loads"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump snaps name=server,type=CephChoices,strings=--server,req=false", + asok_hook, + "dump snapshots"); + ceph_assert(r == 0); + r = admin_socket->register_command("session ls " + "name=cap_dump,type=CephBool,req=false " + "name=filters,type=CephString,n=N,req=false ", + asok_hook, + "List client sessions based on a filter"); + ceph_assert(r == 0); + r = admin_socket->register_command("client ls " + "name=cap_dump,type=CephBool,req=false " + "name=filters,type=CephString,n=N,req=false ", + asok_hook, + "List client sessions based on a filter"); + ceph_assert(r == 0); + r = admin_socket->register_command("session evict name=filters,type=CephString,n=N,req=false", + asok_hook, + "Evict client session(s) based on a filter"); + ceph_assert(r == 0); + r = admin_socket->register_command("client evict name=filters,type=CephString,n=N,req=false", + asok_hook, + "Evict client session(s) based on a filter"); + ceph_assert(r == 0); + r = admin_socket->register_command("session kill name=client_id,type=CephString", + asok_hook, + "Evict a client session by id"); + ceph_assert(r == 0); + r = admin_socket->register_command("session config " + "name=client_id,type=CephInt,req=true " + "name=option,type=CephString,req=true " + "name=value,type=CephString,req=false ", + asok_hook, + "Config a CephFS client session"); + ceph_assert(r == 0); + r = admin_socket->register_command("client config " + "name=client_id,type=CephInt,req=true " + "name=option,type=CephString,req=true " + "name=value,type=CephString,req=false ", + asok_hook, + "Config a CephFS client session"); + ceph_assert(r == 0); + r = admin_socket->register_command("damage ls", + asok_hook, + "List detected metadata damage"); + ceph_assert(r == 0); + r = admin_socket->register_command("damage rm " + "name=damage_id,type=CephInt", + asok_hook, + "Remove a damage table entry"); + ceph_assert(r == 0); + r = admin_socket->register_command("osdmap barrier name=target_epoch,type=CephInt", + asok_hook, + "Wait until the MDS has this OSD map epoch"); + ceph_assert(r == 0); + r = admin_socket->register_command("flush journal", + asok_hook, + "Flush the journal to the backing store"); + ceph_assert(r == 0); + r = admin_socket->register_command("force_readonly", + asok_hook, + "Force MDS to read-only mode"); + ceph_assert(r == 0); + r = admin_socket->register_command("get subtrees", + asok_hook, + "Return the subtree map"); + ceph_assert(r == 0); + r = admin_socket->register_command("dirfrag split " + "name=path,type=CephString,req=true " + "name=frag,type=CephString,req=true " + "name=bits,type=CephInt,req=true ", + asok_hook, + "Fragment directory by path"); + ceph_assert(r == 0); + r = admin_socket->register_command("dirfrag merge " + "name=path,type=CephString,req=true " + "name=frag,type=CephString,req=true", + asok_hook, + "De-fragment directory by path"); + ceph_assert(r == 0); + r = admin_socket->register_command("dirfrag ls " + "name=path,type=CephString,req=true", + asok_hook, + "List fragments in directory"); + ceph_assert(r == 0); + r = admin_socket->register_command("openfiles ls", + asok_hook, + "List the opening files and their caps"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump inode " + "name=number,type=CephInt,req=true", + asok_hook, + "dump inode by inode number"); + ceph_assert(r == 0); + r = admin_socket->register_command("exit", + asok_hook, + "Terminate this MDS"); + r = admin_socket->register_command("respawn", + asok_hook, + "Respawn this MDS"); + ceph_assert(r == 0); + r = admin_socket->register_command( + "heap " \ + "name=heapcmd,type=CephChoices,strings=" \ + "dump|start_profiler|stop_profiler|release|get_release_rate|set_release_rate|stats " \ + "name=value,type=CephString,req=false", + asok_hook, + "show heap usage info (available only if compiled with tcmalloc)"); + ceph_assert(r == 0); + r = admin_socket->register_command( + "cpu_profiler " \ + "name=arg,type=CephChoices,strings=status|flush", + asok_hook, + "run cpu profiling on daemon"); + ceph_assert(r == 0); +} + +void MDSDaemon::clean_up_admin_socket() +{ + g_ceph_context->get_admin_socket()->unregister_commands(asok_hook); + delete asok_hook; + asok_hook = NULL; +} + +int MDSDaemon::init() +{ +#ifdef _WIN32 + // Some file related flags and types are stubbed on Windows. In order to avoid + // incorrect behavior, we're going to prevent the MDS from running on Windows + // until those limitations are addressed. MDS clients, however, are allowed + // to run on Windows. + derr << "The Ceph MDS does not support running on Windows at the moment." + << dendl; + return -CEPHFS_ENOSYS; +#endif // _WIN32 + + dout(10) << "Dumping misc struct sizes:" << dendl; + dout(10) << sizeof(MDSCacheObject) << "\tMDSCacheObject" << dendl; + dout(10) << sizeof(CInode) << "\tCInode" << dendl; + dout(10) << sizeof(elist<void*>::item) << "\telist<>::item" << dendl; + dout(10) << sizeof(CInode::mempool_inode) << "\tinode" << dendl; + dout(10) << sizeof(CInode::mempool_old_inode) << "\told_inode" << dendl; + dout(10) << sizeof(nest_info_t) << "\tnest_info_t" << dendl; + dout(10) << sizeof(frag_info_t) << "\tfrag_info_t" << dendl; + dout(10) << sizeof(SimpleLock) << "\tSimpleLock" << dendl; + dout(10) << sizeof(ScatterLock) << "\tScatterLock" << dendl; + dout(10) << sizeof(CDentry) << "\tCDentry" << dendl; + dout(10) << sizeof(elist<void*>::item) << "\telist<>::item" << dendl; + dout(10) << sizeof(SimpleLock) << "\tSimpleLock" << dendl; + dout(10) << sizeof(CDir) << "\tCDir" << dendl; + dout(10) << sizeof(elist<void*>::item) << "\telist<>::item" << dendl; + dout(10) << sizeof(fnode_t) << "\tfnode_t" << dendl; + dout(10) << sizeof(nest_info_t) << "\tnest_info_t" << dendl; + dout(10) << sizeof(frag_info_t) << "\tfrag_info_t" << dendl; + dout(10) << sizeof(Capability) << "\tCapability" << dendl; + dout(10) << sizeof(xlist<void*>::item) << "\txlist<>::item" << dendl; + + messenger->add_dispatcher_tail(&beacon); + messenger->add_dispatcher_tail(this); + + // init monc + monc->set_messenger(messenger); + + monc->set_want_keys(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD | + CEPH_ENTITY_TYPE_MDS | CEPH_ENTITY_TYPE_MGR); + int r = 0; + r = monc->init(); + if (r < 0) { + derr << "ERROR: failed to init monc: " << cpp_strerror(-r) << dendl; + mds_lock.lock(); + suicide(); + mds_lock.unlock(); + return r; + } + + messenger->set_auth_client(monc); + messenger->set_auth_server(monc); + monc->set_handle_authentication_dispatcher(this); + + // tell monc about log_client so it will know about mon session resets + monc->set_log_client(&log_client); + + r = monc->authenticate(); + if (r < 0) { + derr << "ERROR: failed to authenticate: " << cpp_strerror(-r) << dendl; + mds_lock.lock(); + suicide(); + mds_lock.unlock(); + return r; + } + + int rotating_auth_attempts = 0; + auto rotating_auth_timeout = + g_conf().get_val<int64_t>("rotating_keys_bootstrap_timeout"); + while (monc->wait_auth_rotating(rotating_auth_timeout) < 0) { + if (++rotating_auth_attempts <= g_conf()->max_rotating_auth_attempts) { + derr << "unable to obtain rotating service keys; retrying" << dendl; + continue; + } + derr << "ERROR: failed to refresh rotating keys, " + << "maximum retry time reached." + << " Maybe I have a clock skew against the monitors?" << dendl; + std::lock_guard locker{mds_lock}; + suicide(); + return -CEPHFS_ETIMEDOUT; + } + + mds_lock.lock(); + if (beacon.get_want_state() == CEPH_MDS_STATE_DNE) { + dout(4) << __func__ << ": terminated already, dropping out" << dendl; + mds_lock.unlock(); + return 0; + } + + monc->sub_want("mdsmap", 0, 0); + monc->renew_subs(); + + mds_lock.unlock(); + + // Set up admin socket before taking mds_lock, so that ordering + // is consistent (later we take mds_lock within asok callbacks) + set_up_admin_socket(); + std::lock_guard locker{mds_lock}; + if (beacon.get_want_state() == MDSMap::STATE_DNE) { + suicide(); // we could do something more graceful here + dout(4) << __func__ << ": terminated already, dropping out" << dendl; + return 0; + } + + timer.init(); + + beacon.init(*mdsmap); + messenger->set_myname(entity_name_t::MDS(MDS_RANK_NONE)); + + // schedule tick + reset_tick(); + return 0; +} + +void MDSDaemon::reset_tick() +{ + // cancel old + if (tick_event) timer.cancel_event(tick_event); + + // schedule + tick_event = timer.add_event_after( + g_conf()->mds_tick_interval, + new LambdaContext([this](int) { + ceph_assert(ceph_mutex_is_locked_by_me(mds_lock)); + tick(); + })); +} + +void MDSDaemon::tick() +{ + // reschedule + reset_tick(); + + // Call through to subsystems' tick functions + if (mds_rank) { + mds_rank->tick(); + } +} + +void MDSDaemon::handle_command(const cref_t<MCommand> &m) +{ + auto priv = m->get_connection()->get_priv(); + auto session = static_cast<Session *>(priv.get()); + ceph_assert(session != NULL); + + int r = 0; + cmdmap_t cmdmap; + CachedStackStringStream css; + auto& ss = *css; + bufferlist outbl; + + // If someone is using a closed session for sending commands (e.g. + // the ceph CLI) then we should feel free to clean up this connection + // as soon as we've sent them a response. + const bool live_session = + session->get_state_seq() > 0 && + mds_rank && + mds_rank->sessionmap.get_session(session->info.inst.name); + + if (!live_session) { + // This session only existed to issue commands, so terminate it + // as soon as we can. + ceph_assert(session->is_closed()); + session->get_connection()->mark_disposable(); + } + priv.reset(); + + if (!session->auth_caps.allow_all()) { + dout(1) << __func__ + << ": received command from client without `tell` capability: " + << *m->get_connection()->peer_addrs << dendl; + + ss << "permission denied"; + r = -CEPHFS_EACCES; + } else if (m->cmd.empty()) { + r = -CEPHFS_EINVAL; + ss << "no command given"; + } else if (!TOPNSPC::common::cmdmap_from_json(m->cmd, &cmdmap, ss)) { + r = -CEPHFS_EINVAL; + } else { + cct->get_admin_socket()->queue_tell_command(m); + return; + } + + auto reply = make_message<MCommandReply>(r, ss.str()); + reply->set_tid(m->get_tid()); + reply->set_data(outbl); + m->get_connection()->send_message2(reply); +} + +void MDSDaemon::handle_mds_map(const cref_t<MMDSMap> &m) +{ + version_t epoch = m->get_epoch(); + + // is it new? + if (epoch <= mdsmap->get_epoch()) { + dout(5) << "handle_mds_map old map epoch " << epoch << " <= " + << mdsmap->get_epoch() << ", discarding" << dendl; + return; + } + + dout(1) << "Updating MDS map to version " << epoch << " from " << m->get_source() << dendl; + + // keep old map, for a moment + std::unique_ptr<MDSMap> oldmap; + oldmap.swap(mdsmap); + + // decode and process + mdsmap.reset(new MDSMap); + mdsmap->decode(m->get_encoded()); + + monc->sub_got("mdsmap", mdsmap->get_epoch()); + + // verify compatset + CompatSet mdsmap_compat(MDSMap::get_compat_set_all()); + dout(10) << " my compat " << mdsmap_compat << dendl; + dout(10) << " mdsmap compat " << mdsmap->compat << dendl; + if (!mdsmap_compat.writeable(mdsmap->compat)) { + dout(0) << "handle_mds_map mdsmap compatset " << mdsmap->compat + << " not writeable with daemon features " << mdsmap_compat + << ", killing myself" << dendl; + suicide(); + return; + } + + // Calculate my effective rank (either my owned rank or the rank I'm following if STATE_STANDBY_REPLAY + const auto addrs = messenger->get_myaddrs(); + const auto myid = monc->get_global_id(); + const auto mygid = mds_gid_t(myid); + const auto whoami = mdsmap->get_rank_gid(mygid); + const auto old_state = oldmap->get_state_gid(mygid); + const auto new_state = mdsmap->get_state_gid(mygid); + const auto incarnation = mdsmap->get_inc_gid(mygid); + dout(10) << "my gid is " << myid << dendl; + dout(10) << "map says I am mds." << whoami << "." << incarnation + << " state " << ceph_mds_state_name(new_state) << dendl; + dout(10) << "msgr says I am " << addrs << dendl; + + // If we're removed from the MDSMap, stop all processing. + using DS = MDSMap::DaemonState; + if (old_state != DS::STATE_NULL && new_state == DS::STATE_NULL) { + const auto& oldinfo = oldmap->get_info_gid(mygid); + dout(1) << "Map removed me " << oldinfo + << " from cluster; respawning! See cluster/monitor logs for details." << dendl; + respawn(); + } + + if (old_state == DS::STATE_NULL && new_state != DS::STATE_NULL) { + /* The MDS has been added to the FSMap, now we can init the MgrClient */ + mgrc.init(); + messenger->add_dispatcher_tail(&mgrc); + monc->sub_want("mgrmap", 0, 0); + monc->renew_subs(); /* MgrMap receipt drives connection to ceph-mgr */ + } + + // mark down any failed peers + for (const auto& [gid, info] : oldmap->get_mds_info()) { + if (mdsmap->get_mds_info().count(gid) == 0) { + dout(10) << " peer mds gid " << gid << " removed from map" << dendl; + messenger->mark_down_addrs(info.addrs); + } + } + + if (whoami == MDS_RANK_NONE) { + // We do not hold a rank: + dout(10) << __func__ << ": handling map in rankless mode" << dendl; + + if (new_state == DS::STATE_STANDBY) { + /* Note: STATE_BOOT is never an actual state in the FSMap. The Monitors + * generally mark a new MDS as STANDBY (although it's possible to + * immediately be assigned a rank). + */ + if (old_state == DS::STATE_NULL) { + dout(1) << "Monitors have assigned me to become a standby." << dendl; + beacon.set_want_state(*mdsmap, new_state); + } else if (old_state == DS::STATE_STANDBY) { + dout(5) << "I am still standby" << dendl; + } + } else if (new_state == DS::STATE_NULL) { + /* We are not in the MDSMap yet! Keep waiting: */ + ceph_assert(beacon.get_want_state() == DS::STATE_BOOT); + dout(10) << "not in map yet" << dendl; + } else { + /* We moved to standby somehow from another state */ + ceph_abort("invalid transition to standby"); + } + } else { + // Did we already hold a different rank? MDSMonitor shouldn't try + // to change that out from under me! + if (mds_rank && whoami != mds_rank->get_nodeid()) { + derr << "Invalid rank transition " << mds_rank->get_nodeid() << "->" + << whoami << dendl; + respawn(); + } + + // Did I previously not hold a rank? Initialize! + if (mds_rank == NULL) { + mds_rank = new MDSRankDispatcher(whoami, mds_lock, clog, + timer, beacon, mdsmap, messenger, monc, &mgrc, + new LambdaContext([this](int r){respawn();}), + new LambdaContext([this](int r){suicide();}), + ioctx); + dout(10) << __func__ << ": initializing MDS rank " + << mds_rank->get_nodeid() << dendl; + mds_rank->init(); + } + + // MDSRank is active: let him process the map, we have no say. + dout(10) << __func__ << ": handling map as rank " + << mds_rank->get_nodeid() << dendl; + mds_rank->handle_mds_map(m, *oldmap); + } + + beacon.notify_mdsmap(*mdsmap); +} + +void MDSDaemon::handle_signal(int signum) +{ + ceph_assert(signum == SIGINT || signum == SIGTERM); + derr << "*** got signal " << sig_str(signum) << " ***" << dendl; + { + std::lock_guard l(mds_lock); + if (stopping) { + return; + } + suicide(); + } +} + +void MDSDaemon::suicide() +{ + ceph_assert(ceph_mutex_is_locked(mds_lock)); + + // make sure we don't suicide twice + ceph_assert(stopping == false); + stopping = true; + + dout(1) << "suicide! Wanted state " + << ceph_mds_state_name(beacon.get_want_state()) << dendl; + + if (tick_event) { + timer.cancel_event(tick_event); + tick_event = 0; + } + + clean_up_admin_socket(); + + // Notify the Monitors (MDSMonitor) that we're dying, so that it doesn't have + // to wait for us to go laggy. Only do this if we're actually in the MDSMap, + // because otherwise the MDSMonitor will drop our message. + beacon.set_want_state(*mdsmap, MDSMap::STATE_DNE); + if (!mdsmap->is_dne_gid(mds_gid_t(monc->get_global_id()))) { + beacon.send_and_wait(1); + } + beacon.shutdown(); + + if (mgrc.is_initialized()) + mgrc.shutdown(); + + if (mds_rank) { + mds_rank->shutdown(); + } else { + timer.shutdown(); + + monc->shutdown(); + messenger->shutdown(); + } +} + +void MDSDaemon::respawn() +{ + // --- WARNING TO FUTURE COPY/PASTERS --- + // You must also add a call like + // + // ceph_pthread_setname(pthread_self(), "ceph-mds"); + // + // to main() so that /proc/$pid/stat field 2 contains "(ceph-mds)" + // instead of "(exe)", so that killall (and log rotation) will work. + + dout(1) << "respawn!" << dendl; + + /* Dump recent in case the MDS was stuck doing something which caused it to + * be removed from the MDSMap leading to respawn. */ + g_ceph_context->_log->dump_recent(); + + /* valgrind can't handle execve; just exit and let QA infra restart */ + if (g_conf().get_val<bool>("mds_valgrind_exit")) { + _exit(0); + } + + char *new_argv[orig_argc+1]; + dout(1) << " e: '" << orig_argv[0] << "'" << dendl; + for (int i=0; i<orig_argc; i++) { + new_argv[i] = (char *)orig_argv[i]; + dout(1) << " " << i << ": '" << orig_argv[i] << "'" << dendl; + } + new_argv[orig_argc] = NULL; + + /* Determine the path to our executable, test if Linux /proc/self/exe exists. + * This allows us to exec the same executable even if it has since been + * unlinked. + */ + char exe_path[PATH_MAX] = ""; +#ifdef PROCPREFIX + if (readlink(PROCPREFIX "/proc/self/exe", exe_path, PATH_MAX-1) != -1) { + dout(1) << "respawning with exe " << exe_path << dendl; + strcpy(exe_path, PROCPREFIX "/proc/self/exe"); + } else { +#else + { +#endif + /* Print CWD for the user's interest */ + char buf[PATH_MAX]; + char *cwd = getcwd(buf, sizeof(buf)); + ceph_assert(cwd); + dout(1) << " cwd " << cwd << dendl; + + /* Fall back to a best-effort: just running in our CWD */ + strncpy(exe_path, orig_argv[0], PATH_MAX-1); + } + + dout(1) << " exe_path " << exe_path << dendl; + + unblock_all_signals(NULL); + execv(exe_path, new_argv); + + dout(0) << "respawn execv " << orig_argv[0] + << " failed with " << cpp_strerror(errno) << dendl; + + // We have to assert out here, because suicide() returns, and callers + // to respawn expect it never to return. + ceph_abort(); +} + + + +bool MDSDaemon::ms_dispatch2(const ref_t<Message> &m) +{ + std::lock_guard l(mds_lock); + if (stopping) { + return false; + } + + // Drop out early if shutting down + if (beacon.get_want_state() == CEPH_MDS_STATE_DNE) { + dout(10) << " stopping, discarding " << *m << dendl; + return true; + } + + // First see if it's a daemon message + const bool handled_core = handle_core_message(m); + if (handled_core) { + return true; + } + + // Not core, try it as a rank message + if (mds_rank) { + return mds_rank->ms_dispatch(m); + } else { + return false; + } +} + +/* + * high priority messages we always process + */ + +#define ALLOW_MESSAGES_FROM(peers) \ + do { \ + if (m->get_connection() && (m->get_connection()->get_peer_type() & (peers)) == 0) { \ + dout(0) << __FILE__ << "." << __LINE__ << ": filtered out request, peer=" \ + << m->get_connection()->get_peer_type() << " allowing=" \ + << #peers << " message=" << *m << dendl; \ + return true; \ + } \ + } while (0) + +bool MDSDaemon::handle_core_message(const cref_t<Message> &m) +{ + switch (m->get_type()) { + case CEPH_MSG_MON_MAP: + ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MON); + break; + + // MDS + case CEPH_MSG_MDS_MAP: + ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_MDS); + handle_mds_map(ref_cast<MMDSMap>(m)); + break; + + case MSG_REMOVE_SNAPS: + ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MON); + mds_rank->snapserver->handle_remove_snaps(ref_cast<MRemoveSnaps>(m)); + break; + + // OSD + case MSG_COMMAND: + handle_command(ref_cast<MCommand>(m)); + break; + case CEPH_MSG_OSD_MAP: + ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD); + + if (mds_rank) { + mds_rank->handle_osd_map(); + } + break; + + case MSG_MON_COMMAND: + ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MON); + clog->warn() << "dropping `mds tell` command from legacy monitor"; + break; + + default: + return false; + } + return true; +} + +void MDSDaemon::ms_handle_connect(Connection *con) +{ +} + +bool MDSDaemon::ms_handle_reset(Connection *con) +{ + if (con->get_peer_type() != CEPH_ENTITY_TYPE_CLIENT) + return false; + + std::lock_guard l(mds_lock); + if (stopping) { + return false; + } + dout(5) << "ms_handle_reset on " << con->get_peer_socket_addr() << dendl; + if (beacon.get_want_state() == CEPH_MDS_STATE_DNE) + return false; + + auto priv = con->get_priv(); + if (auto session = static_cast<Session *>(priv.get()); session) { + if (session->is_closed()) { + dout(3) << "ms_handle_reset closing connection for session " << session->info.inst << dendl; + con->mark_down(); + con->set_priv(nullptr); + } + } else { + con->mark_down(); + } + return false; +} + + +void MDSDaemon::ms_handle_remote_reset(Connection *con) +{ + if (con->get_peer_type() != CEPH_ENTITY_TYPE_CLIENT) + return; + + std::lock_guard l(mds_lock); + if (stopping) { + return; + } + + dout(5) << "ms_handle_remote_reset on " << con->get_peer_socket_addr() << dendl; + if (beacon.get_want_state() == CEPH_MDS_STATE_DNE) + return; + + auto priv = con->get_priv(); + if (auto session = static_cast<Session *>(priv.get()); session) { + if (session->is_closed()) { + dout(3) << "ms_handle_remote_reset closing connection for session " << session->info.inst << dendl; + con->mark_down(); + con->set_priv(nullptr); + } + } +} + +bool MDSDaemon::ms_handle_refused(Connection *con) +{ + // do nothing for now + return false; +} + +bool MDSDaemon::parse_caps(const AuthCapsInfo& info, MDSAuthCaps& caps) +{ + caps.clear(); + if (info.allow_all) { + caps.set_allow_all(); + return true; + } else { + auto it = info.caps.begin(); + string auth_cap_str; + try { + decode(auth_cap_str, it); + } catch (const buffer::error& e) { + dout(1) << __func__ << ": cannot decode auth caps buffer of length " << info.caps.length() << dendl; + return false; + } + + dout(10) << __func__ << ": parsing auth_cap_str='" << auth_cap_str << "'" << dendl; + CachedStackStringStream cs; + if (caps.parse(auth_cap_str, cs.get())) { + return true; + } else { + dout(1) << __func__ << ": auth cap parse error: " << cs->strv() << " parsing '" << auth_cap_str << "'" << dendl; + return false; + } + } +} + +int MDSDaemon::ms_handle_fast_authentication(Connection *con) +{ + /* N.B. without mds_lock! */ + MDSAuthCaps caps; + return parse_caps(con->get_peer_caps_info(), caps) ? 0 : -1; +} + +void MDSDaemon::ms_handle_accept(Connection *con) +{ + entity_name_t n(con->get_peer_type(), con->get_peer_global_id()); + std::lock_guard l(mds_lock); + if (stopping) { + return; + } + + // We allow connections and assign Session instances to connections + // even if we have not been assigned a rank, because clients with + // "allow *" are allowed to connect and do 'tell' operations before + // we have a rank. + Session *s = NULL; + if (mds_rank) { + // If we do hold a rank, see if this is an existing client establishing + // a new connection, rather than a new client + s = mds_rank->sessionmap.get_session(n); + } + + // Wire up a Session* to this connection + // It doesn't go into a SessionMap instance until it sends an explicit + // request to open a session (initial state of Session is `closed`) + if (!s) { + s = new Session(con); + dout(10) << " new session " << s << " for " << s->info.inst + << " con " << con << dendl; + con->set_priv(RefCountedPtr{s, false}); + if (mds_rank) { + mds_rank->kick_waiters_for_any_client_connection(); + } + } else { + dout(10) << " existing session " << s << " for " << s->info.inst + << " existing con " << s->get_connection() + << ", new/authorizing con " << con << dendl; + con->set_priv(RefCountedPtr{s}); + } + + parse_caps(con->get_peer_caps_info(), s->auth_caps); + + dout(10) << "ms_handle_accept " << con->get_peer_socket_addr() << " con " << con << " session " << s << dendl; + if (s) { + if (s->get_connection() != con) { + dout(10) << " session connection " << s->get_connection() + << " -> " << con << dendl; + s->set_connection(con); + + // send out any queued messages + while (!s->preopen_out_queue.empty()) { + con->send_message2(s->preopen_out_queue.front()); + s->preopen_out_queue.pop_front(); + } + } + } +} + +bool MDSDaemon::is_clean_shutdown() +{ + if (mds_rank) { + return mds_rank->is_stopped(); + } else { + return true; + } +} |