diff options
Diffstat (limited to 'src/mon/Monitor.cc')
| -rw-r--r-- | src/mon/Monitor.cc | 6505 |
1 files changed, 6505 insertions, 0 deletions
diff --git a/src/mon/Monitor.cc b/src/mon/Monitor.cc new file mode 100644 index 00000000..edf3b789 --- /dev/null +++ b/src/mon/Monitor.cc @@ -0,0 +1,6505 @@ +// -*- 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 <iterator> +#include <sstream> +#include <tuple> +#include <stdlib.h> +#include <signal.h> +#include <limits.h> +#include <cstring> +#include <boost/scope_exit.hpp> +#include <boost/algorithm/string/predicate.hpp> + +#include "json_spirit/json_spirit_reader.h" +#include "json_spirit/json_spirit_writer.h" + +#include "Monitor.h" +#include "common/version.h" +#include "common/blkdev.h" +#include "common/cmdparse.h" +#include "common/signal.h" + +#include "osd/OSDMap.h" + +#include "MonitorDBStore.h" + +#include "messages/PaxosServiceMessage.h" +#include "messages/MMonMap.h" +#include "messages/MMonGetMap.h" +#include "messages/MMonGetVersion.h" +#include "messages/MMonGetVersionReply.h" +#include "messages/MGenericMessage.h" +#include "messages/MMonCommand.h" +#include "messages/MMonCommandAck.h" +#include "messages/MMonMetadata.h" +#include "messages/MMonSync.h" +#include "messages/MMonScrub.h" +#include "messages/MMonProbe.h" +#include "messages/MMonJoin.h" +#include "messages/MMonPaxos.h" +#include "messages/MRoute.h" +#include "messages/MForward.h" + +#include "messages/MMonSubscribe.h" +#include "messages/MMonSubscribeAck.h" + +#include "messages/MAuthReply.h" + +#include "messages/MTimeCheck2.h" +#include "messages/MPing.h" + +#include "common/strtol.h" +#include "common/ceph_argparse.h" +#include "common/Timer.h" +#include "common/Clock.h" +#include "common/errno.h" +#include "common/perf_counters.h" +#include "common/admin_socket.h" +#include "global/signal_handler.h" +#include "common/Formatter.h" +#include "include/stringify.h" +#include "include/color.h" +#include "include/ceph_fs.h" +#include "include/str_list.h" + +#include "OSDMonitor.h" +#include "MDSMonitor.h" +#include "MonmapMonitor.h" +#include "LogMonitor.h" +#include "AuthMonitor.h" +#include "MgrMonitor.h" +#include "MgrStatMonitor.h" +#include "ConfigMonitor.h" +#include "mon/QuorumService.h" +#include "mon/HealthMonitor.h" +#include "mon/ConfigKeyService.h" +#include "common/config.h" +#include "common/cmdparse.h" +#include "include/ceph_assert.h" +#include "include/compat.h" +#include "perfglue/heap_profiler.h" + +#include "auth/none/AuthNoneClientHandler.h" + +#define dout_subsys ceph_subsys_mon +#undef dout_prefix +#define dout_prefix _prefix(_dout, this) +static ostream& _prefix(std::ostream *_dout, const Monitor *mon) { + return *_dout << "mon." << mon->name << "@" << mon->rank + << "(" << mon->get_state_name() << ") e" << mon->monmap->get_epoch() << " "; +} + +const string Monitor::MONITOR_NAME = "monitor"; +const string Monitor::MONITOR_STORE_PREFIX = "monitor_store"; + + +#undef FLAG +#undef COMMAND +#undef COMMAND_WITH_FLAG +#define FLAG(f) (MonCommand::FLAG_##f) +#define COMMAND(parsesig, helptext, modulename, req_perms) \ + {parsesig, helptext, modulename, req_perms, FLAG(NONE)}, +#define COMMAND_WITH_FLAG(parsesig, helptext, modulename, req_perms, flags) \ + {parsesig, helptext, modulename, req_perms, flags}, +MonCommand mon_commands[] = { +#include <mon/MonCommands.h> +}; +#undef COMMAND +#undef COMMAND_WITH_FLAG + + + +void C_MonContext::finish(int r) { + if (mon->is_shutdown()) + return; + FunctionContext::finish(r); +} + +Monitor::Monitor(CephContext* cct_, string nm, MonitorDBStore *s, + Messenger *m, Messenger *mgr_m, MonMap *map) : + Dispatcher(cct_), + AuthServer(cct_), + name(nm), + rank(-1), + messenger(m), + con_self(m ? m->get_loopback_connection() : NULL), + lock("Monitor::lock"), + timer(cct_, lock), + finisher(cct_, "mon_finisher", "fin"), + cpu_tp(cct, "Monitor::cpu_tp", "cpu_tp", g_conf()->mon_cpu_threads), + has_ever_joined(false), + logger(NULL), cluster_logger(NULL), cluster_logger_registered(false), + monmap(map), + log_client(cct_, messenger, monmap, LogClient::FLAG_MON), + key_server(cct, &keyring), + auth_cluster_required(cct, + cct->_conf->auth_supported.empty() ? + cct->_conf->auth_cluster_required : cct->_conf->auth_supported), + auth_service_required(cct, + cct->_conf->auth_supported.empty() ? + cct->_conf->auth_service_required : cct->_conf->auth_supported), + mgr_messenger(mgr_m), + mgr_client(cct_, mgr_m), + gss_ktfile_client(cct->_conf.get_val<std::string>("gss_ktab_client_file")), + store(s), + + elector(this), + required_features(0), + leader(0), + quorum_con_features(0), + // scrub + scrub_version(0), + scrub_event(NULL), + scrub_timeout_event(NULL), + + // sync state + sync_provider_count(0), + sync_cookie(0), + sync_full(false), + sync_start_version(0), + sync_timeout_event(NULL), + sync_last_committed_floor(0), + + timecheck_round(0), + timecheck_acks(0), + timecheck_rounds_since_clean(0), + timecheck_event(NULL), + + paxos_service(PAXOS_NUM), + admin_hook(NULL), + routed_request_tid(0), + op_tracker(cct, g_conf().get_val<bool>("mon_enable_op_tracker"), 1) +{ + clog = log_client.create_channel(CLOG_CHANNEL_CLUSTER); + audit_clog = log_client.create_channel(CLOG_CHANNEL_AUDIT); + + update_log_clients(); + + 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); + } + + op_tracker.set_complaint_and_threshold( + g_conf().get_val<std::chrono::seconds>("mon_op_complaint_time").count(), + g_conf().get_val<int64_t>("mon_op_log_threshold")); + op_tracker.set_history_size_and_duration( + g_conf().get_val<uint64_t>("mon_op_history_size"), + g_conf().get_val<std::chrono::seconds>("mon_op_history_duration").count()); + op_tracker.set_history_slow_op_size_and_threshold( + g_conf().get_val<uint64_t>("mon_op_history_slow_op_size"), + g_conf().get_val<std::chrono::seconds>("mon_op_history_slow_op_threshold").count()); + + paxos = new Paxos(this, "paxos"); + + paxos_service[PAXOS_MDSMAP].reset(new MDSMonitor(this, paxos, "mdsmap")); + paxos_service[PAXOS_MONMAP].reset(new MonmapMonitor(this, paxos, "monmap")); + paxos_service[PAXOS_OSDMAP].reset(new OSDMonitor(cct, this, paxos, "osdmap")); + paxos_service[PAXOS_LOG].reset(new LogMonitor(this, paxos, "logm")); + paxos_service[PAXOS_AUTH].reset(new AuthMonitor(this, paxos, "auth")); + paxos_service[PAXOS_MGR].reset(new MgrMonitor(this, paxos, "mgr")); + paxos_service[PAXOS_MGRSTAT].reset(new MgrStatMonitor(this, paxos, "mgrstat")); + paxos_service[PAXOS_HEALTH].reset(new HealthMonitor(this, paxos, "health")); + paxos_service[PAXOS_CONFIG].reset(new ConfigMonitor(this, paxos, "config")); + + config_key_service = new ConfigKeyService(this, paxos); + + bool r = mon_caps.parse("allow *", NULL); + ceph_assert(r); + + exited_quorum = ceph_clock_now(); + + // prepare local commands + local_mon_commands.resize(std::size(mon_commands)); + for (unsigned i = 0; i < std::size(mon_commands); ++i) { + local_mon_commands[i] = mon_commands[i]; + } + MonCommand::encode_vector(local_mon_commands, local_mon_commands_bl); + + prenautilus_local_mon_commands = local_mon_commands; + for (auto& i : prenautilus_local_mon_commands) { + std::string n = cmddesc_get_prenautilus_compat(i.cmdstring); + if (n != i.cmdstring) { + dout(20) << " pre-nautilus cmd " << i.cmdstring << " -> " << n << dendl; + i.cmdstring = n; + } + } + MonCommand::encode_vector(prenautilus_local_mon_commands, prenautilus_local_mon_commands_bl); + + // assume our commands until we have an election. this only means + // we won't reply with EINVAL before the election; any command that + // actually matters will wait until we have quorum etc and then + // retry (and revalidate). + leader_mon_commands = local_mon_commands; +} + +Monitor::~Monitor() +{ + op_tracker.on_shutdown(); + + paxos_service.clear(); + delete config_key_service; + delete paxos; + ceph_assert(session_map.sessions.empty()); +} + + +class AdminHook : public AdminSocketHook { + Monitor *mon; +public: + explicit AdminHook(Monitor *m) : mon(m) {} + bool call(std::string_view command, const cmdmap_t& cmdmap, + std::string_view format, bufferlist& out) override { + stringstream ss; + mon->do_admin_command(command, cmdmap, format, ss); + out.append(ss); + return true; + } +}; + +void Monitor::do_admin_command(std::string_view command, const cmdmap_t& cmdmap, + std::string_view format, std::ostream& ss) +{ + std::lock_guard l(lock); + + boost::scoped_ptr<Formatter> f(Formatter::create(format)); + + string args; + for (auto p = cmdmap.begin(); + p != cmdmap.end(); ++p) { + if (p->first == "prefix") + continue; + if (!args.empty()) + args += ", "; + args += cmd_vartype_stringify(p->second); + } + args = "[" + args + "]"; + + bool read_only = (command == "mon_status" || + command == "mon metadata" || + command == "quorum_status" || + command == "ops" || + command == "sessions"); + + (read_only ? audit_clog->debug() : audit_clog->info()) + << "from='admin socket' entity='admin socket' " + << "cmd='" << command << "' args=" << args << ": dispatch"; + + if (command == "mon_status") { + get_mon_status(f.get(), ss); + if (f) + f->flush(ss); + } else if (command == "quorum_status") { + _quorum_status(f.get(), ss); + } else if (command == "sync_force") { + string validate; + if ((!cmd_getval(g_ceph_context, cmdmap, "validate", validate)) || + (validate != "--yes-i-really-mean-it")) { + ss << "are you SURE? this will mean the monitor store will be erased " + "the next time the monitor is restarted. pass " + "'--yes-i-really-mean-it' if you really do."; + goto abort; + } + sync_force(f.get(), ss); + } else if (command.compare(0, 23, "add_bootstrap_peer_hint") == 0 || + command.compare(0, 24, "add_bootstrap_peer_hintv") == 0) { + if (!_add_bootstrap_peer_hint(command, cmdmap, ss)) + goto abort; + } else if (command == "quorum enter") { + elector.start_participating(); + start_election(); + ss << "started responding to quorum, initiated new election"; + } else if (command == "quorum exit") { + start_election(); + elector.stop_participating(); + ss << "stopped responding to quorum, initiated new election"; + } else if (command == "ops") { + (void)op_tracker.dump_ops_in_flight(f.get()); + if (f) { + f->flush(ss); + } + } else if (command == "sessions") { + + if (f) { + f->open_array_section("sessions"); + for (auto p : session_map.sessions) { + f->dump_object("session", *p); + } + f->close_section(); + f->flush(ss); + } + + } else if (command == "dump_historic_ops") { + if (op_tracker.dump_historic_ops(f.get())) { + f->flush(ss); + } else { + ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \ + please enable \"mon_enable_op_tracker\", and the tracker will start to track new ops received afterwards."; + } + } else if (command == "dump_historic_ops_by_duration" ) { + if (op_tracker.dump_historic_ops(f.get(), true)) { + f->flush(ss); + } else { + ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \ + please enable \"mon_enable_op_tracker\", and the tracker will start to track new ops received afterwards."; + } + } else if (command == "dump_historic_slow_ops") { + if (op_tracker.dump_historic_slow_ops(f.get(), {})) { + f->flush(ss); + } else { + ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \ + please enable \"mon_enable_op_tracker\", and the tracker will start to track new ops received afterwards."; + } + } else { + ceph_abort_msg("bad AdminSocket command binding"); + } + (read_only ? audit_clog->debug() : audit_clog->info()) + << "from='admin socket' " + << "entity='admin socket' " + << "cmd=" << command << " " + << "args=" << args << ": finished"; + return; + +abort: + (read_only ? audit_clog->debug() : audit_clog->info()) + << "from='admin socket' " + << "entity='admin socket' " + << "cmd=" << command << " " + << "args=" << args << ": aborted"; +} + +void Monitor::handle_signal(int signum) +{ + ceph_assert(signum == SIGINT || signum == SIGTERM); + derr << "*** Got Signal " << sig_str(signum) << " ***" << dendl; + shutdown(); +} + +CompatSet Monitor::get_initial_supported_features() +{ + CompatSet::FeatureSet ceph_mon_feature_compat; + CompatSet::FeatureSet ceph_mon_feature_ro_compat; + CompatSet::FeatureSet ceph_mon_feature_incompat; + ceph_mon_feature_incompat.insert(CEPH_MON_FEATURE_INCOMPAT_BASE); + ceph_mon_feature_incompat.insert(CEPH_MON_FEATURE_INCOMPAT_SINGLE_PAXOS); + return CompatSet(ceph_mon_feature_compat, ceph_mon_feature_ro_compat, + ceph_mon_feature_incompat); +} + +CompatSet Monitor::get_supported_features() +{ + CompatSet compat = get_initial_supported_features(); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_OSD_ERASURE_CODES); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_OSDMAP_ENC); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_ERASURE_CODE_PLUGINS_V2); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_ERASURE_CODE_PLUGINS_V3); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_KRAKEN); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_LUMINOUS); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_MIMIC); + compat.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_NAUTILUS); + return compat; +} + +CompatSet Monitor::get_legacy_features() +{ + CompatSet::FeatureSet ceph_mon_feature_compat; + CompatSet::FeatureSet ceph_mon_feature_ro_compat; + CompatSet::FeatureSet ceph_mon_feature_incompat; + ceph_mon_feature_incompat.insert(CEPH_MON_FEATURE_INCOMPAT_BASE); + return CompatSet(ceph_mon_feature_compat, ceph_mon_feature_ro_compat, + ceph_mon_feature_incompat); +} + +int Monitor::check_features(MonitorDBStore *store) +{ + CompatSet required = get_supported_features(); + CompatSet ondisk; + + read_features_off_disk(store, &ondisk); + + if (!required.writeable(ondisk)) { + CompatSet diff = required.unsupported(ondisk); + generic_derr << "ERROR: on disk data includes unsupported features: " << diff << dendl; + return -EPERM; + } + + return 0; +} + +void Monitor::read_features_off_disk(MonitorDBStore *store, CompatSet *features) +{ + bufferlist featuresbl; + store->get(MONITOR_NAME, COMPAT_SET_LOC, featuresbl); + if (featuresbl.length() == 0) { + generic_dout(0) << "WARNING: mon fs missing feature list.\n" + << "Assuming it is old-style and introducing one." << dendl; + //we only want the baseline ~v.18 features assumed to be on disk. + //If new features are introduced this code needs to disappear or + //be made smarter. + *features = get_legacy_features(); + + features->encode(featuresbl); + auto t(std::make_shared<MonitorDBStore::Transaction>()); + t->put(MONITOR_NAME, COMPAT_SET_LOC, featuresbl); + store->apply_transaction(t); + } else { + auto it = featuresbl.cbegin(); + features->decode(it); + } +} + +void Monitor::read_features() +{ + read_features_off_disk(store, &features); + dout(10) << "features " << features << dendl; + + calc_quorum_requirements(); + dout(10) << "required_features " << required_features << dendl; +} + +void Monitor::write_features(MonitorDBStore::TransactionRef t) +{ + bufferlist bl; + features.encode(bl); + t->put(MONITOR_NAME, COMPAT_SET_LOC, bl); +} + +const char** Monitor::get_tracked_conf_keys() const +{ + static const char* KEYS[] = { + "crushtool", // helpful for testing + "mon_election_timeout", + "mon_lease", + "mon_lease_renew_interval_factor", + "mon_lease_ack_timeout_factor", + "mon_accept_timeout_factor", + // clog & admin clog + "clog_to_monitors", + "clog_to_syslog", + "clog_to_syslog_facility", + "clog_to_syslog_level", + "clog_to_graylog", + "clog_to_graylog_host", + "clog_to_graylog_port", + "host", + "fsid", + // periodic health to clog + "mon_health_to_clog", + "mon_health_to_clog_interval", + "mon_health_to_clog_tick_interval", + // scrub interval + "mon_scrub_interval", + "mon_allow_pool_delete", + // osdmap pruning - observed, not handled. + "mon_osdmap_full_prune_enabled", + "mon_osdmap_full_prune_min", + "mon_osdmap_full_prune_interval", + "mon_osdmap_full_prune_txsize", + // debug options - observed, not handled + "mon_debug_extra_checks", + "mon_debug_block_osdmap_trim", + NULL + }; + return KEYS; +} + +void Monitor::handle_conf_change(const ConfigProxy& conf, + const std::set<std::string> &changed) +{ + sanitize_options(); + + dout(10) << __func__ << " " << changed << dendl; + + if (changed.count("clog_to_monitors") || + changed.count("clog_to_syslog") || + changed.count("clog_to_syslog_level") || + changed.count("clog_to_syslog_facility") || + changed.count("clog_to_graylog") || + changed.count("clog_to_graylog_host") || + changed.count("clog_to_graylog_port") || + changed.count("host") || + changed.count("fsid")) { + update_log_clients(); + } + + if (changed.count("mon_health_to_clog") || + changed.count("mon_health_to_clog_interval") || + changed.count("mon_health_to_clog_tick_interval")) { + std::set<std::string> c2(changed); + finisher.queue(new C_MonContext(this, [this, c2](int) { + Mutex::Locker l(lock); + health_to_clog_update_conf(c2); + })); + } + + if (changed.count("mon_scrub_interval")) { + int scrub_interval = conf->mon_scrub_interval; + finisher.queue(new C_MonContext(this, [this, scrub_interval](int) { + Mutex::Locker l(lock); + scrub_update_interval(scrub_interval); + })); + } +} + +void Monitor::update_log_clients() +{ + map<string,string> log_to_monitors; + map<string,string> log_to_syslog; + map<string,string> log_channel; + map<string,string> log_prio; + map<string,string> log_to_graylog; + map<string,string> log_to_graylog_host; + map<string,string> log_to_graylog_port; + uuid_d fsid; + string host; + + if (parse_log_client_options(g_ceph_context, log_to_monitors, log_to_syslog, + log_channel, log_prio, log_to_graylog, + log_to_graylog_host, log_to_graylog_port, + fsid, host)) + return; + + clog->update_config(log_to_monitors, log_to_syslog, + log_channel, log_prio, log_to_graylog, + log_to_graylog_host, log_to_graylog_port, + fsid, host); + + audit_clog->update_config(log_to_monitors, log_to_syslog, + log_channel, log_prio, log_to_graylog, + log_to_graylog_host, log_to_graylog_port, + fsid, host); +} + +int Monitor::sanitize_options() +{ + int r = 0; + + // mon_lease must be greater than mon_lease_renewal; otherwise we + // may incur in leases expiring before they are renewed. + if (g_conf()->mon_lease_renew_interval_factor >= 1.0) { + clog->error() << "mon_lease_renew_interval_factor (" + << g_conf()->mon_lease_renew_interval_factor + << ") must be less than 1.0"; + r = -EINVAL; + } + + // mon_lease_ack_timeout must be greater than mon_lease to make sure we've + // got time to renew the lease and get an ack for it. Having both options + // with the same value, for a given small vale, could mean timing out if + // the monitors happened to be overloaded -- or even under normal load for + // a small enough value. + if (g_conf()->mon_lease_ack_timeout_factor <= 1.0) { + clog->error() << "mon_lease_ack_timeout_factor (" + << g_conf()->mon_lease_ack_timeout_factor + << ") must be greater than 1.0"; + r = -EINVAL; + } + + return r; +} + +int Monitor::preinit() +{ + lock.Lock(); + + dout(1) << "preinit fsid " << monmap->fsid << dendl; + + int r = sanitize_options(); + if (r < 0) { + derr << "option sanitization failed!" << dendl; + lock.Unlock(); + return r; + } + + ceph_assert(!logger); + { + PerfCountersBuilder pcb(g_ceph_context, "mon", l_mon_first, l_mon_last); + pcb.add_u64(l_mon_num_sessions, "num_sessions", "Open sessions", "sess", + PerfCountersBuilder::PRIO_USEFUL); + pcb.add_u64_counter(l_mon_session_add, "session_add", "Created sessions", + "sadd", PerfCountersBuilder::PRIO_INTERESTING); + pcb.add_u64_counter(l_mon_session_rm, "session_rm", "Removed sessions", + "srm", PerfCountersBuilder::PRIO_INTERESTING); + pcb.add_u64_counter(l_mon_session_trim, "session_trim", "Trimmed sessions", + "strm", PerfCountersBuilder::PRIO_USEFUL); + pcb.add_u64_counter(l_mon_num_elections, "num_elections", "Elections participated in", + "ecnt", PerfCountersBuilder::PRIO_USEFUL); + pcb.add_u64_counter(l_mon_election_call, "election_call", "Elections started", + "estt", PerfCountersBuilder::PRIO_INTERESTING); + pcb.add_u64_counter(l_mon_election_win, "election_win", "Elections won", + "ewon", PerfCountersBuilder::PRIO_INTERESTING); + pcb.add_u64_counter(l_mon_election_lose, "election_lose", "Elections lost", + "elst", PerfCountersBuilder::PRIO_INTERESTING); + logger = pcb.create_perf_counters(); + cct->get_perfcounters_collection()->add(logger); + } + + ceph_assert(!cluster_logger); + { + PerfCountersBuilder pcb(g_ceph_context, "cluster", l_cluster_first, l_cluster_last); + pcb.add_u64(l_cluster_num_mon, "num_mon", "Monitors"); + pcb.add_u64(l_cluster_num_mon_quorum, "num_mon_quorum", "Monitors in quorum"); + pcb.add_u64(l_cluster_num_osd, "num_osd", "OSDs"); + pcb.add_u64(l_cluster_num_osd_up, "num_osd_up", "OSDs that are up"); + pcb.add_u64(l_cluster_num_osd_in, "num_osd_in", "OSD in state \"in\" (they are in cluster)"); + pcb.add_u64(l_cluster_osd_epoch, "osd_epoch", "Current epoch of OSD map"); + pcb.add_u64(l_cluster_osd_bytes, "osd_bytes", "Total capacity of cluster", NULL, 0, unit_t(UNIT_BYTES)); + pcb.add_u64(l_cluster_osd_bytes_used, "osd_bytes_used", "Used space", NULL, 0, unit_t(UNIT_BYTES)); + pcb.add_u64(l_cluster_osd_bytes_avail, "osd_bytes_avail", "Available space", NULL, 0, unit_t(UNIT_BYTES)); + pcb.add_u64(l_cluster_num_pool, "num_pool", "Pools"); + pcb.add_u64(l_cluster_num_pg, "num_pg", "Placement groups"); + pcb.add_u64(l_cluster_num_pg_active_clean, "num_pg_active_clean", "Placement groups in active+clean state"); + pcb.add_u64(l_cluster_num_pg_active, "num_pg_active", "Placement groups in active state"); + pcb.add_u64(l_cluster_num_pg_peering, "num_pg_peering", "Placement groups in peering state"); + pcb.add_u64(l_cluster_num_object, "num_object", "Objects"); + pcb.add_u64(l_cluster_num_object_degraded, "num_object_degraded", "Degraded (missing replicas) objects"); + pcb.add_u64(l_cluster_num_object_misplaced, "num_object_misplaced", "Misplaced (wrong location in the cluster) objects"); + pcb.add_u64(l_cluster_num_object_unfound, "num_object_unfound", "Unfound objects"); + pcb.add_u64(l_cluster_num_bytes, "num_bytes", "Size of all objects", NULL, 0, unit_t(UNIT_BYTES)); + cluster_logger = pcb.create_perf_counters(); + } + + paxos->init_logger(); + + // verify cluster_uuid + { + int r = check_fsid(); + if (r == -ENOENT) + r = write_fsid(); + if (r < 0) { + lock.Unlock(); + return r; + } + } + + // open compatset + read_features(); + + // have we ever joined a quorum? + has_ever_joined = (store->get(MONITOR_NAME, "joined") != 0); + dout(10) << "has_ever_joined = " << (int)has_ever_joined << dendl; + + if (!has_ever_joined) { + // impose initial quorum restrictions? + list<string> initial_members; + get_str_list(g_conf()->mon_initial_members, initial_members); + + if (!initial_members.empty()) { + dout(1) << " initial_members " << initial_members << ", filtering seed monmap" << dendl; + + monmap->set_initial_members( + g_ceph_context, initial_members, name, messenger->get_myaddrs(), + &extra_probe_peers); + + dout(10) << " monmap is " << *monmap << dendl; + dout(10) << " extra probe peers " << extra_probe_peers << dendl; + } + } else if (!monmap->contains(name)) { + derr << "not in monmap and have been in a quorum before; " + << "must have been removed" << dendl; + if (g_conf()->mon_force_quorum_join) { + dout(0) << "we should have died but " + << "'mon_force_quorum_join' is set -- allowing boot" << dendl; + } else { + derr << "commit suicide!" << dendl; + lock.Unlock(); + return -ENOENT; + } + } + + { + // We have a potentially inconsistent store state in hands. Get rid of it + // and start fresh. + bool clear_store = false; + if (store->exists("mon_sync", "in_sync")) { + dout(1) << __func__ << " clean up potentially inconsistent store state" + << dendl; + clear_store = true; + } + + if (store->get("mon_sync", "force_sync") > 0) { + dout(1) << __func__ << " force sync by clearing store state" << dendl; + clear_store = true; + } + + if (clear_store) { + set<string> sync_prefixes = get_sync_targets_names(); + store->clear(sync_prefixes); + } + } + + sync_last_committed_floor = store->get("mon_sync", "last_committed_floor"); + dout(10) << "sync_last_committed_floor " << sync_last_committed_floor << dendl; + + init_paxos(); + + if (is_keyring_required()) { + // we need to bootstrap authentication keys so we can form an + // initial quorum. + if (authmon()->get_last_committed() == 0) { + dout(10) << "loading initial keyring to bootstrap authentication for mkfs" << dendl; + bufferlist bl; + int err = store->get("mkfs", "keyring", bl); + if (err == 0 && bl.length() > 0) { + // Attempt to decode and extract keyring only if it is found. + KeyRing keyring; + auto p = bl.cbegin(); + decode(keyring, p); + extract_save_mon_key(keyring); + } + } + + string keyring_loc = g_conf()->mon_data + "/keyring"; + + r = keyring.load(cct, keyring_loc); + if (r < 0) { + EntityName mon_name; + mon_name.set_type(CEPH_ENTITY_TYPE_MON); + EntityAuth mon_key; + if (key_server.get_auth(mon_name, mon_key)) { + dout(1) << "copying mon. key from old db to external keyring" << dendl; + keyring.add(mon_name, mon_key); + bufferlist bl; + keyring.encode_plaintext(bl); + write_default_keyring(bl); + } else { + derr << "unable to load initial keyring " << g_conf()->keyring << dendl; + lock.Unlock(); + return r; + } + } + } + + admin_hook = new AdminHook(this); + AdminSocket* admin_socket = cct->get_admin_socket(); + + // unlock while registering to avoid mon_lock -> admin socket lock dependency. + lock.Unlock(); + r = admin_socket->register_command("mon_status", "mon_status", admin_hook, + "show current monitor status"); + ceph_assert(r == 0); + r = admin_socket->register_command("quorum_status", "quorum_status", + admin_hook, "show current quorum status"); + ceph_assert(r == 0); + r = admin_socket->register_command("sync_force", + "sync_force name=validate," + "type=CephChoices," + "strings=--yes-i-really-mean-it", + admin_hook, + "force sync of and clear monitor store"); + ceph_assert(r == 0); + r = admin_socket->register_command("add_bootstrap_peer_hint", + "add_bootstrap_peer_hint name=addr," + "type=CephIPAddr", + admin_hook, + "add peer address as potential bootstrap" + " peer for cluster bringup"); + ceph_assert(r == 0); + r = admin_socket->register_command("add_bootstrap_peer_hintv", + "add_bootstrap_peer_hintv name=addrv," + "type=CephString", + admin_hook, + "add peer address vector as potential bootstrap" + " peer for cluster bringup"); + ceph_assert(r == 0); + r = admin_socket->register_command("quorum enter", "quorum enter", + admin_hook, + "force monitor back into quorum"); + ceph_assert(r == 0); + r = admin_socket->register_command("quorum exit", "quorum exit", + admin_hook, + "force monitor out of the quorum"); + ceph_assert(r == 0); + r = admin_socket->register_command("ops", + "ops", + admin_hook, + "show the ops currently in flight"); + ceph_assert(r == 0); + r = admin_socket->register_command("sessions", + "sessions", + admin_hook, + "list existing sessions"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_historic_ops", "dump_historic_ops", + admin_hook, + "show recent ops"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_historic_ops_by_duration", "dump_historic_ops_by_duration", + admin_hook, + "show recent ops, sorted by duration"); + ceph_assert(r == 0); + r = admin_socket->register_command("dump_historic_slow_ops", "dump_historic_slow_ops", + admin_hook, + "show recent slow ops"); + ceph_assert(r == 0); + + lock.Lock(); + + // add ourselves as a conf observer + g_conf().add_observer(this); + + messenger->set_auth_client(this); + messenger->set_auth_server(this); + mgr_messenger->set_auth_client(this); + + auth_registry.refresh_config(); + + lock.Unlock(); + return 0; +} + +int Monitor::init() +{ + dout(2) << "init" << dendl; + std::lock_guard l(lock); + + finisher.start(); + + // start ticker + timer.init(); + new_tick(); + + cpu_tp.start(); + + // i'm ready! + messenger->add_dispatcher_tail(this); + + // kickstart pet mgrclient + mgr_client.init(); + mgr_messenger->add_dispatcher_tail(&mgr_client); + mgr_messenger->add_dispatcher_tail(this); // for auth ms_* calls + mgrmon()->prime_mgr_client(); + + state = STATE_PROBING; + bootstrap(); + // add features of myself into feature_map + session_map.feature_map.add_mon(con_self->get_features()); + return 0; +} + +void Monitor::init_paxos() +{ + dout(10) << __func__ << dendl; + paxos->init(); + + // init services + for (auto& svc : paxos_service) { + svc->init(); + } + + refresh_from_paxos(NULL); +} + +void Monitor::refresh_from_paxos(bool *need_bootstrap) +{ + dout(10) << __func__ << dendl; + + bufferlist bl; + int r = store->get(MONITOR_NAME, "cluster_fingerprint", bl); + if (r >= 0) { + try { + auto p = bl.cbegin(); + decode(fingerprint, p); + } + catch (buffer::error& e) { + dout(10) << __func__ << " failed to decode cluster_fingerprint" << dendl; + } + } else { + dout(10) << __func__ << " no cluster_fingerprint" << dendl; + } + + for (auto& svc : paxos_service) { + svc->refresh(need_bootstrap); + } + for (auto& svc : paxos_service) { + svc->post_refresh(); + } + load_metadata(); +} + +void Monitor::register_cluster_logger() +{ + if (!cluster_logger_registered) { + dout(10) << "register_cluster_logger" << dendl; + cluster_logger_registered = true; + cct->get_perfcounters_collection()->add(cluster_logger); + } else { + dout(10) << "register_cluster_logger - already registered" << dendl; + } +} + +void Monitor::unregister_cluster_logger() +{ + if (cluster_logger_registered) { + dout(10) << "unregister_cluster_logger" << dendl; + cluster_logger_registered = false; + cct->get_perfcounters_collection()->remove(cluster_logger); + } else { + dout(10) << "unregister_cluster_logger - not registered" << dendl; + } +} + +void Monitor::update_logger() +{ + cluster_logger->set(l_cluster_num_mon, monmap->size()); + cluster_logger->set(l_cluster_num_mon_quorum, quorum.size()); +} + +void Monitor::shutdown() +{ + dout(1) << "shutdown" << dendl; + + lock.Lock(); + + wait_for_paxos_write(); + + { + std::lock_guard l(auth_lock); + authmon()->_set_mon_num_rank(0, 0); + } + + state = STATE_SHUTDOWN; + + lock.Unlock(); + g_conf().remove_observer(this); + lock.Lock(); + + if (admin_hook) { + cct->get_admin_socket()->unregister_commands(admin_hook); + delete admin_hook; + admin_hook = NULL; + } + + elector.shutdown(); + + mgr_client.shutdown(); + + lock.Unlock(); + finisher.wait_for_empty(); + finisher.stop(); + lock.Lock(); + + // clean up + paxos->shutdown(); + for (auto& svc : paxos_service) { + svc->shutdown(); + } + + finish_contexts(g_ceph_context, waitfor_quorum, -ECANCELED); + finish_contexts(g_ceph_context, maybe_wait_for_quorum, -ECANCELED); + + timer.shutdown(); + + cpu_tp.stop(); + + remove_all_sessions(); + + log_client.shutdown(); + + // unlock before msgr shutdown... + lock.Unlock(); + + // shutdown messenger before removing logger from perfcounter collection, + // otherwise _ms_dispatch() will try to update deleted logger + messenger->shutdown(); + mgr_messenger->shutdown(); + + if (logger) { + cct->get_perfcounters_collection()->remove(logger); + delete logger; + logger = NULL; + } + if (cluster_logger) { + if (cluster_logger_registered) + cct->get_perfcounters_collection()->remove(cluster_logger); + delete cluster_logger; + cluster_logger = NULL; + } +} + +void Monitor::wait_for_paxos_write() +{ + if (paxos->is_writing() || paxos->is_writing_previous()) { + dout(10) << __func__ << " flushing pending write" << dendl; + lock.Unlock(); + store->flush(); + lock.Lock(); + dout(10) << __func__ << " flushed pending write" << dendl; + } +} + +void Monitor::respawn() +{ + // --- WARNING TO FUTURE COPY/PASTERS --- + // You must also add a call like + // + // ceph_pthread_setname(pthread_self(), "ceph-mon"); + // + // to main() so that /proc/$pid/stat field 2 contains "(ceph-mon)" + // instead of "(exe)", so that killall (and log rotation) will work. + + dout(0) << __func__ << dendl; + + 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(); +} + +void Monitor::bootstrap() +{ + dout(10) << "bootstrap" << dendl; + wait_for_paxos_write(); + + sync_reset_requester(); + unregister_cluster_logger(); + cancel_probe_timeout(); + + if (monmap->get_epoch() == 0) { + dout(10) << "reverting to legacy ranks for seed monmap (epoch 0)" << dendl; + monmap->calc_legacy_ranks(); + } + dout(10) << "monmap " << *monmap << dendl; + + if (monmap->min_mon_release && + monmap->min_mon_release + 2 < (int)ceph_release()) { + derr << "current monmap has min_mon_release " + << (int)monmap->min_mon_release + << " (" << ceph_release_name(monmap->min_mon_release) + << ") which is >2 releases older than me " << ceph_release() + << " (" << ceph_release_name(ceph_release()) << "), stopping." + << dendl; + exit(0); + } + + // note my rank + int newrank = monmap->get_rank(messenger->get_myaddrs()); + if (newrank < 0 && rank >= 0) { + // was i ever part of the quorum? + if (has_ever_joined) { + dout(0) << " removed from monmap, suicide." << dendl; + exit(0); + } + } + if (newrank >= 0 && + monmap->get_addrs(newrank) != messenger->get_myaddrs()) { + dout(0) << " monmap addrs for rank " << newrank << " changed, i am " + << messenger->get_myaddrs() + << ", monmap is " << monmap->get_addrs(newrank) << ", respawning" + << dendl; + + if (monmap->get_epoch()) { + // store this map in temp mon_sync location so that we use it on + // our next startup + derr << " stashing newest monmap " << monmap->get_epoch() + << " for next startup" << dendl; + bufferlist bl; + monmap->encode(bl, -1); + auto t(std::make_shared<MonitorDBStore::Transaction>()); + t->put("mon_sync", "temp_newer_monmap", bl); + store->apply_transaction(t); + } + + respawn(); + } + if (newrank != rank) { + dout(0) << " my rank is now " << newrank << " (was " << rank << ")" << dendl; + messenger->set_myname(entity_name_t::MON(newrank)); + rank = newrank; + + // reset all connections, or else our peers will think we are someone else. + messenger->mark_down_all(); + } + + // reset + state = STATE_PROBING; + + _reset(); + + // sync store + if (g_conf()->mon_compact_on_bootstrap) { + dout(10) << "bootstrap -- triggering compaction" << dendl; + store->compact(); + dout(10) << "bootstrap -- finished compaction" << dendl; + } + + // singleton monitor? + if (monmap->size() == 1 && rank == 0) { + win_standalone_election(); + return; + } + + reset_probe_timeout(); + + // i'm outside the quorum + if (monmap->contains(name)) + outside_quorum.insert(name); + + // probe monitors + dout(10) << "probing other monitors" << dendl; + for (unsigned i = 0; i < monmap->size(); i++) { + if ((int)i != rank) + send_mon_message( + new MMonProbe(monmap->fsid, MMonProbe::OP_PROBE, name, has_ever_joined, + ceph_release()), + i); + } + for (auto& av : extra_probe_peers) { + if (av != messenger->get_myaddrs()) { + messenger->send_to_mon( + new MMonProbe(monmap->fsid, MMonProbe::OP_PROBE, name, has_ever_joined, + ceph_release()), + av); + } + } +} + +bool Monitor::_add_bootstrap_peer_hint(std::string_view cmd, + const cmdmap_t& cmdmap, + ostream& ss) +{ + if (is_leader() || is_peon()) { + ss << "mon already active; ignoring bootstrap hint"; + return true; + } + + entity_addrvec_t addrs; + string addrstr; + if (cmd_getval(g_ceph_context, cmdmap, "addr", addrstr)) { + dout(10) << "_add_bootstrap_peer_hint '" << cmd << "' addr '" + << addrstr << "'" << dendl; + + entity_addr_t addr; + const char *end = 0; + if (!addr.parse(addrstr.c_str(), &end, entity_addr_t::TYPE_ANY)) { + ss << "failed to parse addrs '" << addrstr + << "'; syntax is 'add_bootstrap_peer_hint ip[:port]'"; + return false; + } + + addrs.v.push_back(addr); + if (addr.get_port() == 0) { + addrs.v[0].set_type(entity_addr_t::TYPE_MSGR2); + addrs.v[0].set_port(CEPH_MON_PORT_IANA); + addrs.v.push_back(addr); + addrs.v[1].set_type(entity_addr_t::TYPE_LEGACY); + addrs.v[1].set_port(CEPH_MON_PORT_LEGACY); + } else if (addr.get_type() == entity_addr_t::TYPE_ANY) { + if (addr.get_port() == CEPH_MON_PORT_LEGACY) { + addrs.v[0].set_type(entity_addr_t::TYPE_LEGACY); + } else { + addrs.v[0].set_type(entity_addr_t::TYPE_MSGR2); + } + } + } else if (cmd_getval(g_ceph_context, cmdmap, "addrv", addrstr)) { + dout(10) << "_add_bootstrap_peer_hintv '" << cmd << "' addrv '" + << addrstr << "'" << dendl; + const char *end = 0; + if (!addrs.parse(addrstr.c_str(), &end)) { + ss << "failed to parse addrs '" << addrstr + << "'; syntax is 'add_bootstrap_peer_hintv v2:ip:port[,v1:ip:port]'"; + return false; + } + } else { + ss << "no addr or addrv provided"; + return false; + } + + extra_probe_peers.insert(addrs); + ss << "adding peer " << addrs << " to list: " << extra_probe_peers; + return true; +} + +// called by bootstrap(), or on leader|peon -> electing +void Monitor::_reset() +{ + dout(10) << __func__ << dendl; + + // disable authentication + { + std::lock_guard l(auth_lock); + authmon()->_set_mon_num_rank(0, 0); + } + + cancel_probe_timeout(); + timecheck_finish(); + health_events_cleanup(); + health_check_log_times.clear(); + scrub_event_cancel(); + + leader_since = utime_t(); + quorum_since = {}; + if (!quorum.empty()) { + exited_quorum = ceph_clock_now(); + } + quorum.clear(); + outside_quorum.clear(); + quorum_feature_map.clear(); + + scrub_reset(); + + paxos->restart(); + + for (auto& svc : paxos_service) { + svc->restart(); + } +} + + +// ----------------------------------------------------------- +// sync + +set<string> Monitor::get_sync_targets_names() +{ + set<string> targets; + targets.insert(paxos->get_name()); + for (auto& svc : paxos_service) { + svc->get_store_prefixes(targets); + } + ConfigKeyService *config_key_service_ptr = dynamic_cast<ConfigKeyService*>(config_key_service); + ceph_assert(config_key_service_ptr); + config_key_service_ptr->get_store_prefixes(targets); + return targets; +} + + +void Monitor::sync_timeout() +{ + dout(10) << __func__ << dendl; + ceph_assert(state == STATE_SYNCHRONIZING); + bootstrap(); +} + +void Monitor::sync_obtain_latest_monmap(bufferlist &bl) +{ + dout(1) << __func__ << dendl; + + MonMap latest_monmap; + + // Grab latest monmap from MonmapMonitor + bufferlist monmon_bl; + int err = monmon()->get_monmap(monmon_bl); + if (err < 0) { + if (err != -ENOENT) { + derr << __func__ + << " something wrong happened while reading the store: " + << cpp_strerror(err) << dendl; + ceph_abort_msg("error reading the store"); + } + } else { + latest_monmap.decode(monmon_bl); + } + + // Grab last backed up monmap (if any) and compare epochs + if (store->exists("mon_sync", "latest_monmap")) { + bufferlist backup_bl; + int err = store->get("mon_sync", "latest_monmap", backup_bl); + if (err < 0) { + derr << __func__ + << " something wrong happened while reading the store: " + << cpp_strerror(err) << dendl; + ceph_abort_msg("error reading the store"); + } + ceph_assert(backup_bl.length() > 0); + + MonMap backup_monmap; + backup_monmap.decode(backup_bl); + + if (backup_monmap.epoch > latest_monmap.epoch) + latest_monmap = backup_monmap; + } + + // Check if our current monmap's epoch is greater than the one we've + // got so far. + if (monmap->epoch > latest_monmap.epoch) + latest_monmap = *monmap; + + dout(1) << __func__ << " obtained monmap e" << latest_monmap.epoch << dendl; + + latest_monmap.encode(bl, CEPH_FEATURES_ALL); +} + +void Monitor::sync_reset_requester() +{ + dout(10) << __func__ << dendl; + + if (sync_timeout_event) { + timer.cancel_event(sync_timeout_event); + sync_timeout_event = NULL; + } + + sync_provider = entity_addrvec_t(); + sync_cookie = 0; + sync_full = false; + sync_start_version = 0; +} + +void Monitor::sync_reset_provider() +{ + dout(10) << __func__ << dendl; + sync_providers.clear(); +} + +void Monitor::sync_start(entity_addrvec_t &addrs, bool full) +{ + dout(10) << __func__ << " " << addrs << (full ? " full" : " recent") << dendl; + + ceph_assert(state == STATE_PROBING || + state == STATE_SYNCHRONIZING); + state = STATE_SYNCHRONIZING; + + // make sure are not a provider for anyone! + sync_reset_provider(); + + sync_full = full; + + if (sync_full) { + // stash key state, and mark that we are syncing + auto t(std::make_shared<MonitorDBStore::Transaction>()); + sync_stash_critical_state(t); + t->put("mon_sync", "in_sync", 1); + + sync_last_committed_floor = std::max(sync_last_committed_floor, paxos->get_version()); + dout(10) << __func__ << " marking sync in progress, storing sync_last_committed_floor " + << sync_last_committed_floor << dendl; + t->put("mon_sync", "last_committed_floor", sync_last_committed_floor); + + store->apply_transaction(t); + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 1); + + // clear the underlying store + set<string> targets = get_sync_targets_names(); + dout(10) << __func__ << " clearing prefixes " << targets << dendl; + store->clear(targets); + + // make sure paxos knows it has been reset. this prevents a + // bootstrap and then different probe reply order from possibly + // deciding a partial or no sync is needed. + paxos->init(); + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 2); + } + + // assume 'other' as the leader. We will update the leader once we receive + // a reply to the sync start. + sync_provider = addrs; + + sync_reset_timeout(); + + MMonSync *m = new MMonSync(sync_full ? MMonSync::OP_GET_COOKIE_FULL : MMonSync::OP_GET_COOKIE_RECENT); + if (!sync_full) + m->last_committed = paxos->get_version(); + messenger->send_to_mon(m, sync_provider); +} + +void Monitor::sync_stash_critical_state(MonitorDBStore::TransactionRef t) +{ + dout(10) << __func__ << dendl; + bufferlist backup_monmap; + sync_obtain_latest_monmap(backup_monmap); + ceph_assert(backup_monmap.length() > 0); + t->put("mon_sync", "latest_monmap", backup_monmap); +} + +void Monitor::sync_reset_timeout() +{ + dout(10) << __func__ << dendl; + if (sync_timeout_event) + timer.cancel_event(sync_timeout_event); + sync_timeout_event = timer.add_event_after( + g_conf()->mon_sync_timeout, + new C_MonContext(this, [this](int) { + sync_timeout(); + })); +} + +void Monitor::sync_finish(version_t last_committed) +{ + dout(10) << __func__ << " lc " << last_committed << " from " << sync_provider << dendl; + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 7); + + if (sync_full) { + // finalize the paxos commits + auto tx(std::make_shared<MonitorDBStore::Transaction>()); + paxos->read_and_prepare_transactions(tx, sync_start_version, + last_committed); + tx->put(paxos->get_name(), "last_committed", last_committed); + + dout(30) << __func__ << " final tx dump:\n"; + JSONFormatter f(true); + tx->dump(&f); + f.flush(*_dout); + *_dout << dendl; + + store->apply_transaction(tx); + } + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 8); + + auto t(std::make_shared<MonitorDBStore::Transaction>()); + t->erase("mon_sync", "in_sync"); + t->erase("mon_sync", "force_sync"); + t->erase("mon_sync", "last_committed_floor"); + store->apply_transaction(t); + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 9); + + init_paxos(); + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 10); + + bootstrap(); +} + +void Monitor::handle_sync(MonOpRequestRef op) +{ + MMonSync *m = static_cast<MMonSync*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + switch (m->op) { + + // provider --------- + + case MMonSync::OP_GET_COOKIE_FULL: + case MMonSync::OP_GET_COOKIE_RECENT: + handle_sync_get_cookie(op); + break; + case MMonSync::OP_GET_CHUNK: + handle_sync_get_chunk(op); + break; + + // client ----------- + + case MMonSync::OP_COOKIE: + handle_sync_cookie(op); + break; + + case MMonSync::OP_CHUNK: + case MMonSync::OP_LAST_CHUNK: + handle_sync_chunk(op); + break; + case MMonSync::OP_NO_COOKIE: + handle_sync_no_cookie(op); + break; + + default: + dout(0) << __func__ << " unknown op " << m->op << dendl; + ceph_abort_msg("unknown op"); + } +} + +// leader + +void Monitor::_sync_reply_no_cookie(MonOpRequestRef op) +{ + MMonSync *m = static_cast<MMonSync*>(op->get_req()); + MMonSync *reply = new MMonSync(MMonSync::OP_NO_COOKIE, m->cookie); + m->get_connection()->send_message(reply); +} + +void Monitor::handle_sync_get_cookie(MonOpRequestRef op) +{ + MMonSync *m = static_cast<MMonSync*>(op->get_req()); + if (is_synchronizing()) { + _sync_reply_no_cookie(op); + return; + } + + ceph_assert(g_conf()->mon_sync_provider_kill_at != 1); + + // make sure they can understand us. + if ((required_features ^ m->get_connection()->get_features()) & + required_features) { + dout(5) << " ignoring peer mon." << m->get_source().num() + << " has features " << std::hex + << m->get_connection()->get_features() + << " but we require " << required_features << std::dec << dendl; + return; + } + + // make up a unique cookie. include election epoch (which persists + // across restarts for the whole cluster) and a counter for this + // process instance. there is no need to be unique *across* + // monitors, though. + uint64_t cookie = ((unsigned long long)elector.get_epoch() << 24) + ++sync_provider_count; + ceph_assert(sync_providers.count(cookie) == 0); + + dout(10) << __func__ << " cookie " << cookie << " for " << m->get_source_inst() << dendl; + + SyncProvider& sp = sync_providers[cookie]; + sp.cookie = cookie; + sp.addrs = m->get_source_addrs(); + sp.reset_timeout(g_ceph_context, g_conf()->mon_sync_timeout * 2); + + set<string> sync_targets; + if (m->op == MMonSync::OP_GET_COOKIE_FULL) { + // full scan + sync_targets = get_sync_targets_names(); + sp.last_committed = paxos->get_version(); + sp.synchronizer = store->get_synchronizer(sp.last_key, sync_targets); + sp.full = true; + dout(10) << __func__ << " will sync prefixes " << sync_targets << dendl; + } else { + // just catch up paxos + sp.last_committed = m->last_committed; + } + dout(10) << __func__ << " will sync from version " << sp.last_committed << dendl; + + MMonSync *reply = new MMonSync(MMonSync::OP_COOKIE, sp.cookie); + reply->last_committed = sp.last_committed; + m->get_connection()->send_message(reply); +} + +void Monitor::handle_sync_get_chunk(MonOpRequestRef op) +{ + MMonSync *m = static_cast<MMonSync*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + + if (sync_providers.count(m->cookie) == 0) { + dout(10) << __func__ << " no cookie " << m->cookie << dendl; + _sync_reply_no_cookie(op); + return; + } + + ceph_assert(g_conf()->mon_sync_provider_kill_at != 2); + + SyncProvider& sp = sync_providers[m->cookie]; + sp.reset_timeout(g_ceph_context, g_conf()->mon_sync_timeout * 2); + + if (sp.last_committed < paxos->get_first_committed() && + paxos->get_first_committed() > 1) { + dout(10) << __func__ << " sync requester fell behind paxos, their lc " << sp.last_committed + << " < our fc " << paxos->get_first_committed() << dendl; + sync_providers.erase(m->cookie); + _sync_reply_no_cookie(op); + return; + } + + MMonSync *reply = new MMonSync(MMonSync::OP_CHUNK, sp.cookie); + auto tx(std::make_shared<MonitorDBStore::Transaction>()); + + int bytes_left = g_conf()->mon_sync_max_payload_size; + int keys_left = g_conf()->mon_sync_max_payload_keys; + while (sp.last_committed < paxos->get_version() && + bytes_left > 0 && + keys_left > 0) { + bufferlist bl; + sp.last_committed++; + + int err = store->get(paxos->get_name(), sp.last_committed, bl); + ceph_assert(err == 0); + + tx->put(paxos->get_name(), sp.last_committed, bl); + bytes_left -= bl.length(); + --keys_left; + dout(20) << __func__ << " including paxos state " << sp.last_committed + << dendl; + } + reply->last_committed = sp.last_committed; + + if (sp.full && bytes_left > 0 && keys_left > 0) { + sp.synchronizer->get_chunk_tx(tx, bytes_left, keys_left); + sp.last_key = sp.synchronizer->get_last_key(); + reply->last_key = sp.last_key; + } + + if ((sp.full && sp.synchronizer->has_next_chunk()) || + sp.last_committed < paxos->get_version()) { + dout(10) << __func__ << " chunk, through version " << sp.last_committed + << " key " << sp.last_key << dendl; + } else { + dout(10) << __func__ << " last chunk, through version " << sp.last_committed + << " key " << sp.last_key << dendl; + reply->op = MMonSync::OP_LAST_CHUNK; + + ceph_assert(g_conf()->mon_sync_provider_kill_at != 3); + + // clean up our local state + sync_providers.erase(sp.cookie); + } + + encode(*tx, reply->chunk_bl); + + m->get_connection()->send_message(reply); +} + +// requester + +void Monitor::handle_sync_cookie(MonOpRequestRef op) +{ + MMonSync *m = static_cast<MMonSync*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + if (sync_cookie) { + dout(10) << __func__ << " already have a cookie, ignoring" << dendl; + return; + } + if (m->get_source_addrs() != sync_provider) { + dout(10) << __func__ << " source does not match, discarding" << dendl; + return; + } + sync_cookie = m->cookie; + sync_start_version = m->last_committed; + + sync_reset_timeout(); + sync_get_next_chunk(); + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 3); +} + +void Monitor::sync_get_next_chunk() +{ + dout(20) << __func__ << " cookie " << sync_cookie << " provider " << sync_provider << dendl; + if (g_conf()->mon_inject_sync_get_chunk_delay > 0) { + dout(20) << __func__ << " injecting delay of " << g_conf()->mon_inject_sync_get_chunk_delay << dendl; + usleep((long long)(g_conf()->mon_inject_sync_get_chunk_delay * 1000000.0)); + } + MMonSync *r = new MMonSync(MMonSync::OP_GET_CHUNK, sync_cookie); + messenger->send_to_mon(r, sync_provider); + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 4); +} + +void Monitor::handle_sync_chunk(MonOpRequestRef op) +{ + MMonSync *m = static_cast<MMonSync*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + + if (m->cookie != sync_cookie) { + dout(10) << __func__ << " cookie does not match, discarding" << dendl; + return; + } + if (m->get_source_addrs() != sync_provider) { + dout(10) << __func__ << " source does not match, discarding" << dendl; + return; + } + + ceph_assert(state == STATE_SYNCHRONIZING); + ceph_assert(g_conf()->mon_sync_requester_kill_at != 5); + + auto tx(std::make_shared<MonitorDBStore::Transaction>()); + tx->append_from_encoded(m->chunk_bl); + + dout(30) << __func__ << " tx dump:\n"; + JSONFormatter f(true); + tx->dump(&f); + f.flush(*_dout); + *_dout << dendl; + + store->apply_transaction(tx); + + ceph_assert(g_conf()->mon_sync_requester_kill_at != 6); + + if (!sync_full) { + dout(10) << __func__ << " applying recent paxos transactions as we go" << dendl; + auto tx(std::make_shared<MonitorDBStore::Transaction>()); + paxos->read_and_prepare_transactions(tx, paxos->get_version() + 1, + m->last_committed); + tx->put(paxos->get_name(), "last_committed", m->last_committed); + + dout(30) << __func__ << " tx dump:\n"; + JSONFormatter f(true); + tx->dump(&f); + f.flush(*_dout); + *_dout << dendl; + + store->apply_transaction(tx); + paxos->init(); // to refresh what we just wrote + } + + if (m->op == MMonSync::OP_CHUNK) { + sync_reset_timeout(); + sync_get_next_chunk(); + } else if (m->op == MMonSync::OP_LAST_CHUNK) { + sync_finish(m->last_committed); + } +} + +void Monitor::handle_sync_no_cookie(MonOpRequestRef op) +{ + dout(10) << __func__ << dendl; + bootstrap(); +} + +void Monitor::sync_trim_providers() +{ + dout(20) << __func__ << dendl; + + utime_t now = ceph_clock_now(); + map<uint64_t,SyncProvider>::iterator p = sync_providers.begin(); + while (p != sync_providers.end()) { + if (now > p->second.timeout) { + dout(10) << __func__ << " expiring cookie " << p->second.cookie + << " for " << p->second.addrs << dendl; + sync_providers.erase(p++); + } else { + ++p; + } + } +} + +// --------------------------------------------------- +// probe + +void Monitor::cancel_probe_timeout() +{ + if (probe_timeout_event) { + dout(10) << "cancel_probe_timeout " << probe_timeout_event << dendl; + timer.cancel_event(probe_timeout_event); + probe_timeout_event = NULL; + } else { + dout(10) << "cancel_probe_timeout (none scheduled)" << dendl; + } +} + +void Monitor::reset_probe_timeout() +{ + cancel_probe_timeout(); + probe_timeout_event = new C_MonContext(this, [this](int r) { + probe_timeout(r); + }); + double t = g_conf()->mon_probe_timeout; + if (timer.add_event_after(t, probe_timeout_event)) { + dout(10) << "reset_probe_timeout " << probe_timeout_event + << " after " << t << " seconds" << dendl; + } else { + probe_timeout_event = nullptr; + } +} + +void Monitor::probe_timeout(int r) +{ + dout(4) << "probe_timeout " << probe_timeout_event << dendl; + ceph_assert(is_probing() || is_synchronizing()); + ceph_assert(probe_timeout_event); + probe_timeout_event = NULL; + bootstrap(); +} + +void Monitor::handle_probe(MonOpRequestRef op) +{ + MMonProbe *m = static_cast<MMonProbe*>(op->get_req()); + dout(10) << "handle_probe " << *m << dendl; + + if (m->fsid != monmap->fsid) { + dout(0) << "handle_probe ignoring fsid " << m->fsid << " != " << monmap->fsid << dendl; + return; + } + + switch (m->op) { + case MMonProbe::OP_PROBE: + handle_probe_probe(op); + break; + + case MMonProbe::OP_REPLY: + handle_probe_reply(op); + break; + + case MMonProbe::OP_MISSING_FEATURES: + derr << __func__ << " require release " << (int)m->mon_release << " > " + << (int)ceph_release() + << ", or missing features (have " << CEPH_FEATURES_ALL + << ", required " << m->required_features + << ", missing " << (m->required_features & ~CEPH_FEATURES_ALL) << ")" + << dendl; + break; + } +} + +void Monitor::handle_probe_probe(MonOpRequestRef op) +{ + MMonProbe *m = static_cast<MMonProbe*>(op->get_req()); + + dout(10) << "handle_probe_probe " << m->get_source_inst() << *m + << " features " << m->get_connection()->get_features() << dendl; + uint64_t missing = required_features & ~m->get_connection()->get_features(); + if ((m->mon_release > 0 && m->mon_release < monmap->min_mon_release) || + missing) { + dout(1) << " peer " << m->get_source_addr() + << " release " << (int)m->mon_release + << " < min_mon_release " << (int)monmap->min_mon_release + << ", or missing features " << missing << dendl; + MMonProbe *r = new MMonProbe(monmap->fsid, MMonProbe::OP_MISSING_FEATURES, + name, has_ever_joined, monmap->min_mon_release); + m->required_features = required_features; + m->get_connection()->send_message(r); + goto out; + } + + if (!is_probing() && !is_synchronizing()) { + // If the probing mon is way ahead of us, we need to re-bootstrap. + // Normally we capture this case when we initially bootstrap, but + // it is possible we pass those checks (we overlap with + // quorum-to-be) but fail to join a quorum before it moves past + // us. We need to be kicked back to bootstrap so we can + // synchonize, not keep calling elections. + if (paxos->get_version() + 1 < m->paxos_first_version) { + dout(1) << " peer " << m->get_source_addr() << " has first_committed " + << "ahead of us, re-bootstrapping" << dendl; + bootstrap(); + goto out; + + } + } + + MMonProbe *r; + r = new MMonProbe(monmap->fsid, MMonProbe::OP_REPLY, name, has_ever_joined, + ceph_release()); + r->name = name; + r->quorum = quorum; + monmap->encode(r->monmap_bl, m->get_connection()->get_features()); + r->paxos_first_version = paxos->get_first_committed(); + r->paxos_last_version = paxos->get_version(); + m->get_connection()->send_message(r); + + // did we discover a peer here? + if (!monmap->contains(m->get_source_addr())) { + dout(1) << " adding peer " << m->get_source_addrs() + << " to list of hints" << dendl; + extra_probe_peers.insert(m->get_source_addrs()); + } + + out: + return; +} + +void Monitor::handle_probe_reply(MonOpRequestRef op) +{ + MMonProbe *m = static_cast<MMonProbe*>(op->get_req()); + dout(10) << "handle_probe_reply " << m->get_source_inst() + << " " << *m << dendl; + dout(10) << " monmap is " << *monmap << dendl; + + // discover name and addrs during probing or electing states. + if (!is_probing() && !is_electing()) { + return; + } + + // newer map, or they've joined a quorum and we haven't? + bufferlist mybl; + monmap->encode(mybl, m->get_connection()->get_features()); + // make sure it's actually different; the checks below err toward + // taking the other guy's map, which could cause us to loop. + if (!mybl.contents_equal(m->monmap_bl)) { + MonMap *newmap = new MonMap; + newmap->decode(m->monmap_bl); + if (m->has_ever_joined && (newmap->get_epoch() > monmap->get_epoch() || + !has_ever_joined)) { + dout(10) << " got newer/committed monmap epoch " << newmap->get_epoch() + << ", mine was " << monmap->get_epoch() << dendl; + delete newmap; + monmap->decode(m->monmap_bl); + + bootstrap(); + return; + } + delete newmap; + } + + // rename peer? + string peer_name = monmap->get_name(m->get_source_addr()); + if (monmap->get_epoch() == 0 && peer_name.compare(0, 7, "noname-") == 0) { + dout(10) << " renaming peer " << m->get_source_addr() << " " + << peer_name << " -> " << m->name << " in my monmap" + << dendl; + monmap->rename(peer_name, m->name); + + if (is_electing()) { + bootstrap(); + return; + } + } else if (peer_name.size()) { + dout(10) << " peer name is " << peer_name << dendl; + } else { + dout(10) << " peer " << m->get_source_addr() << " not in map" << dendl; + } + + // new initial peer? + if (monmap->get_epoch() == 0 && + monmap->contains(m->name) && + monmap->get_addrs(m->name).front().is_blank_ip()) { + dout(1) << " learned initial mon " << m->name + << " addrs " << m->get_source_addrs() << dendl; + monmap->set_addrvec(m->name, m->get_source_addrs()); + + bootstrap(); + return; + } + + // end discover phase + if (!is_probing()) { + return; + } + + ceph_assert(paxos != NULL); + + if (is_synchronizing()) { + dout(10) << " currently syncing" << dendl; + return; + } + + entity_addrvec_t other = m->get_source_addrs(); + + if (m->paxos_last_version < sync_last_committed_floor) { + dout(10) << " peer paxos versions [" << m->paxos_first_version + << "," << m->paxos_last_version << "] < my sync_last_committed_floor " + << sync_last_committed_floor << ", ignoring" + << dendl; + } else { + if (paxos->get_version() < m->paxos_first_version && + m->paxos_first_version > 1) { // no need to sync if we're 0 and they start at 1. + dout(10) << " peer paxos first versions [" << m->paxos_first_version + << "," << m->paxos_last_version << "]" + << " vs my version " << paxos->get_version() + << " (too far ahead)" + << dendl; + cancel_probe_timeout(); + sync_start(other, true); + return; + } + if (paxos->get_version() + g_conf()->paxos_max_join_drift < m->paxos_last_version) { + dout(10) << " peer paxos last version " << m->paxos_last_version + << " vs my version " << paxos->get_version() + << " (too far ahead)" + << dendl; + cancel_probe_timeout(); + sync_start(other, false); + return; + } + } + + // did the existing cluster complete upgrade to luminous? + if (osdmon()->osdmap.get_epoch()) { + if (osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) { + derr << __func__ << " existing cluster has not completed upgrade to" + << " luminous; 'ceph osd require_osd_release luminous' before" + << " upgrading" << dendl; + exit(0); + } + if (!osdmon()->osdmap.test_flag(CEPH_OSDMAP_PURGED_SNAPDIRS) || + !osdmon()->osdmap.test_flag(CEPH_OSDMAP_RECOVERY_DELETES)) { + derr << __func__ << " existing cluster has not completed a full luminous" + << " scrub to purge legacy snapdir objects; please scrub before" + << " upgrading beyond luminous." << dendl; + exit(0); + } + } + + // is there an existing quorum? + if (m->quorum.size()) { + dout(10) << " existing quorum " << m->quorum << dendl; + + dout(10) << " peer paxos version " << m->paxos_last_version + << " vs my version " << paxos->get_version() + << " (ok)" + << dendl; + + if (monmap->contains(name) && + !monmap->get_addrs(name).front().is_blank_ip()) { + // i'm part of the cluster; just initiate a new election + start_election(); + } else { + dout(10) << " ready to join, but i'm not in the monmap or my addr is blank, trying to join" << dendl; + send_mon_message( + new MMonJoin(monmap->fsid, name, messenger->get_myaddrs()), + *m->quorum.begin()); + } + } else { + if (monmap->contains(m->name)) { + dout(10) << " mon." << m->name << " is outside the quorum" << dendl; + outside_quorum.insert(m->name); + } else { + dout(10) << " mostly ignoring mon." << m->name << ", not part of monmap" << dendl; + return; + } + + unsigned need = monmap->min_quorum_size(); + dout(10) << " outside_quorum now " << outside_quorum << ", need " << need << dendl; + if (outside_quorum.size() >= need) { + if (outside_quorum.count(name)) { + dout(10) << " that's enough to form a new quorum, calling election" << dendl; + start_election(); + } else { + dout(10) << " that's enough to form a new quorum, but it does not include me; waiting" << dendl; + } + } else { + dout(10) << " that's not yet enough for a new quorum, waiting" << dendl; + } + } +} + +void Monitor::join_election() +{ + dout(10) << __func__ << dendl; + wait_for_paxos_write(); + _reset(); + state = STATE_ELECTING; + + logger->inc(l_mon_num_elections); +} + +void Monitor::start_election() +{ + dout(10) << "start_election" << dendl; + wait_for_paxos_write(); + _reset(); + state = STATE_ELECTING; + + logger->inc(l_mon_num_elections); + logger->inc(l_mon_election_call); + + clog->info() << "mon." << name << " calling monitor election"; + elector.call_election(); +} + +void Monitor::win_standalone_election() +{ + dout(1) << "win_standalone_election" << dendl; + + // bump election epoch, in case the previous epoch included other + // monitors; we need to be able to make the distinction. + elector.init(); + elector.advance_epoch(); + + rank = monmap->get_rank(name); + ceph_assert(rank == 0); + set<int> q; + q.insert(rank); + + map<int,Metadata> metadata; + collect_metadata(&metadata[0]); + + win_election(elector.get_epoch(), q, + CEPH_FEATURES_ALL, + ceph::features::mon::get_supported(), + ceph_release(), + metadata); +} + +const utime_t& Monitor::get_leader_since() const +{ + ceph_assert(state == STATE_LEADER); + return leader_since; +} + +epoch_t Monitor::get_epoch() +{ + return elector.get_epoch(); +} + +void Monitor::_finish_svc_election() +{ + ceph_assert(state == STATE_LEADER || state == STATE_PEON); + + for (auto& svc : paxos_service) { + // we already called election_finished() on monmon(); avoid callig twice + if (state == STATE_LEADER && svc.get() == monmon()) + continue; + svc->election_finished(); + } +} + +void Monitor::win_election(epoch_t epoch, set<int>& active, uint64_t features, + const mon_feature_t& mon_features, + int min_mon_release, + const map<int,Metadata>& metadata) +{ + dout(10) << __func__ << " epoch " << epoch << " quorum " << active + << " features " << features + << " mon_features " << mon_features + << " min_mon_release " << min_mon_release + << dendl; + ceph_assert(is_electing()); + state = STATE_LEADER; + leader_since = ceph_clock_now(); + quorum_since = mono_clock::now(); + leader = rank; + quorum = active; + quorum_con_features = features; + quorum_mon_features = mon_features; + quorum_min_mon_release = min_mon_release; + pending_metadata = metadata; + outside_quorum.clear(); + + clog->info() << "mon." << name << " is new leader, mons " << get_quorum_names() + << " in quorum (ranks " << quorum << ")"; + + set_leader_commands(get_local_commands(mon_features)); + + paxos->leader_init(); + // NOTE: tell monmap monitor first. This is important for the + // bootstrap case to ensure that the very first paxos proposal + // codifies the monmap. Otherwise any manner of chaos can ensue + // when monitors are call elections or participating in a paxos + // round without agreeing on who the participants are. + monmon()->election_finished(); + _finish_svc_election(); + + logger->inc(l_mon_election_win); + + // inject new metadata in first transaction. + { + // include previous metadata for missing mons (that aren't part of + // the current quorum). + map<int,Metadata> m = metadata; + for (unsigned rank = 0; rank < monmap->size(); ++rank) { + if (m.count(rank) == 0 && + mon_metadata.count(rank)) { + m[rank] = mon_metadata[rank]; + } + } + + // FIXME: This is a bit sloppy because we aren't guaranteed to submit + // a new transaction immediately after the election finishes. We should + // do that anyway for other reasons, though. + MonitorDBStore::TransactionRef t = paxos->get_pending_transaction(); + bufferlist bl; + encode(m, bl); + t->put(MONITOR_STORE_PREFIX, "last_metadata", bl); + } + + finish_election(); + if (monmap->size() > 1 && + monmap->get_epoch() > 0) { + timecheck_start(); + health_tick_start(); + + // Freshen the health status before doing health_to_clog in case + // our just-completed election changed the health + healthmon()->wait_for_active_ctx(new FunctionContext([this](int r){ + dout(20) << "healthmon now active" << dendl; + healthmon()->tick(); + if (healthmon()->is_proposing()) { + dout(20) << __func__ << " healthmon proposing, waiting" << dendl; + healthmon()->wait_for_finished_proposal(nullptr, new C_MonContext(this, + [this](int r){ + ceph_assert(lock.is_locked_by_me()); + do_health_to_clog_interval(); + })); + + } else { + do_health_to_clog_interval(); + } + })); + + scrub_event_start(); + } +} + +void Monitor::lose_election(epoch_t epoch, set<int> &q, int l, + uint64_t features, + const mon_feature_t& mon_features, + int min_mon_release) +{ + state = STATE_PEON; + leader_since = utime_t(); + quorum_since = mono_clock::now(); + leader = l; + quorum = q; + outside_quorum.clear(); + quorum_con_features = features; + quorum_mon_features = mon_features; + quorum_min_mon_release = min_mon_release; + dout(10) << "lose_election, epoch " << epoch << " leader is mon" << leader + << " quorum is " << quorum << " features are " << quorum_con_features + << " mon_features are " << quorum_mon_features + << " min_mon_release " << min_mon_release + << dendl; + + paxos->peon_init(); + _finish_svc_election(); + + logger->inc(l_mon_election_lose); + + finish_election(); +} + +namespace { +std::string collect_compression_algorithms() +{ + ostringstream os; + bool printed = false; + for (auto [name, key] : Compressor::compression_algorithms) { + if (printed) { + os << ", "; + } else { + printed = true; + } + std::ignore = key; + os << name; + } + return os.str(); +} +} + +void Monitor::collect_metadata(Metadata *m) +{ + collect_sys_info(m, g_ceph_context); + (*m)["addrs"] = stringify(messenger->get_myaddrs()); + (*m)["compression_algorithms"] = collect_compression_algorithms(); + + // infer storage device + string devname = store->get_devname(); + set<string> devnames; + get_raw_devices(devname, &devnames); + map<string,string> errs; + get_device_metadata(devnames, m, &errs); + for (auto& i : errs) { + dout(1) << __func__ << " " << i.first << ": " << i.second << dendl; + } +} + +void Monitor::finish_election() +{ + apply_quorum_to_compatset_features(); + apply_monmap_to_compatset_features(); + timecheck_finish(); + exited_quorum = utime_t(); + finish_contexts(g_ceph_context, waitfor_quorum); + finish_contexts(g_ceph_context, maybe_wait_for_quorum); + resend_routed_requests(); + update_logger(); + register_cluster_logger(); + + // enable authentication + { + std::lock_guard l(auth_lock); + authmon()->_set_mon_num_rank(monmap->size(), rank); + } + + // am i named properly? + string cur_name = monmap->get_name(messenger->get_myaddrs()); + if (cur_name != name) { + dout(10) << " renaming myself from " << cur_name << " -> " << name << dendl; + send_mon_message( + new MMonJoin(monmap->fsid, name, messenger->get_myaddrs()), + *quorum.begin()); + } +} + +void Monitor::_apply_compatset_features(CompatSet &new_features) +{ + if (new_features.compare(features) != 0) { + CompatSet diff = features.unsupported(new_features); + dout(1) << __func__ << " enabling new quorum features: " << diff << dendl; + features = new_features; + + auto t = std::make_shared<MonitorDBStore::Transaction>(); + write_features(t); + store->apply_transaction(t); + + calc_quorum_requirements(); + } +} + +void Monitor::apply_quorum_to_compatset_features() +{ + CompatSet new_features(features); + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_OSD_ERASURE_CODES); + if (quorum_con_features & CEPH_FEATURE_OSDMAP_ENC) { + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_OSDMAP_ENC); + } + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_ERASURE_CODE_PLUGINS_V2); + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_ERASURE_CODE_PLUGINS_V3); + dout(5) << __func__ << dendl; + _apply_compatset_features(new_features); +} + +void Monitor::apply_monmap_to_compatset_features() +{ + CompatSet new_features(features); + mon_feature_t monmap_features = monmap->get_required_features(); + + /* persistent monmap features may go into the compatset. + * optional monmap features may not - why? + * because optional monmap features may be set/unset by the admin, + * and possibly by other means that haven't yet been thought out, + * so we can't make the monitor enforce them on start - because they + * may go away. + * this, of course, does not invalidate setting a compatset feature + * for an optional feature - as long as you make sure to clean it up + * once you unset it. + */ + if (monmap_features.contains_all(ceph::features::mon::FEATURE_KRAKEN)) { + ceph_assert(ceph::features::mon::get_persistent().contains_all( + ceph::features::mon::FEATURE_KRAKEN)); + // this feature should only ever be set if the quorum supports it. + ceph_assert(HAVE_FEATURE(quorum_con_features, SERVER_KRAKEN)); + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_KRAKEN); + } + if (monmap_features.contains_all(ceph::features::mon::FEATURE_LUMINOUS)) { + ceph_assert(ceph::features::mon::get_persistent().contains_all( + ceph::features::mon::FEATURE_LUMINOUS)); + // this feature should only ever be set if the quorum supports it. + ceph_assert(HAVE_FEATURE(quorum_con_features, SERVER_LUMINOUS)); + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_LUMINOUS); + } + if (monmap_features.contains_all(ceph::features::mon::FEATURE_MIMIC)) { + ceph_assert(ceph::features::mon::get_persistent().contains_all( + ceph::features::mon::FEATURE_MIMIC)); + // this feature should only ever be set if the quorum supports it. + ceph_assert(HAVE_FEATURE(quorum_con_features, SERVER_MIMIC)); + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_MIMIC); + } + if (monmap_features.contains_all(ceph::features::mon::FEATURE_NAUTILUS)) { + ceph_assert(ceph::features::mon::get_persistent().contains_all( + ceph::features::mon::FEATURE_NAUTILUS)); + // this feature should only ever be set if the quorum supports it. + ceph_assert(HAVE_FEATURE(quorum_con_features, SERVER_NAUTILUS)); + new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_NAUTILUS); + } + + dout(5) << __func__ << dendl; + _apply_compatset_features(new_features); +} + +void Monitor::calc_quorum_requirements() +{ + required_features = 0; + + // compatset + if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_OSDMAP_ENC)) { + required_features |= CEPH_FEATURE_OSDMAP_ENC; + } + if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_KRAKEN)) { + required_features |= CEPH_FEATUREMASK_SERVER_KRAKEN; + } + if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_LUMINOUS)) { + required_features |= CEPH_FEATUREMASK_SERVER_LUMINOUS; + } + if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_MIMIC)) { + required_features |= CEPH_FEATUREMASK_SERVER_MIMIC; + } + if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_NAUTILUS)) { + required_features |= CEPH_FEATUREMASK_SERVER_NAUTILUS | + CEPH_FEATUREMASK_CEPHX_V2; + } + + // monmap + if (monmap->get_required_features().contains_all( + ceph::features::mon::FEATURE_KRAKEN)) { + required_features |= CEPH_FEATUREMASK_SERVER_KRAKEN; + } + if (monmap->get_required_features().contains_all( + ceph::features::mon::FEATURE_LUMINOUS)) { + required_features |= CEPH_FEATUREMASK_SERVER_LUMINOUS; + } + if (monmap->get_required_features().contains_all( + ceph::features::mon::FEATURE_MIMIC)) { + required_features |= CEPH_FEATUREMASK_SERVER_MIMIC; + } + if (monmap->get_required_features().contains_all( + ceph::features::mon::FEATURE_NAUTILUS)) { + required_features |= CEPH_FEATUREMASK_SERVER_NAUTILUS | + CEPH_FEATUREMASK_CEPHX_V2; + } + dout(10) << __func__ << " required_features " << required_features << dendl; +} + +void Monitor::get_combined_feature_map(FeatureMap *fm) +{ + *fm += session_map.feature_map; + for (auto id : quorum) { + if (id != rank) { + *fm += quorum_feature_map[id]; + } + } +} + +void Monitor::sync_force(Formatter *f, ostream& ss) +{ + bool free_formatter = false; + + if (!f) { + // louzy/lazy hack: default to json if no formatter has been defined + f = new JSONFormatter(); + free_formatter = true; + } + + auto tx(std::make_shared<MonitorDBStore::Transaction>()); + sync_stash_critical_state(tx); + tx->put("mon_sync", "force_sync", 1); + store->apply_transaction(tx); + + f->open_object_section("sync_force"); + f->dump_int("ret", 0); + f->dump_stream("msg") << "forcing store sync the next time the monitor starts"; + f->close_section(); // sync_force + f->flush(ss); + if (free_formatter) + delete f; +} + +void Monitor::_quorum_status(Formatter *f, ostream& ss) +{ + bool free_formatter = false; + + if (!f) { + // louzy/lazy hack: default to json if no formatter has been defined + f = new JSONFormatter(); + free_formatter = true; + } + f->open_object_section("quorum_status"); + f->dump_int("election_epoch", get_epoch()); + + f->open_array_section("quorum"); + for (set<int>::iterator p = quorum.begin(); p != quorum.end(); ++p) + f->dump_int("mon", *p); + f->close_section(); // quorum + + list<string> quorum_names = get_quorum_names(); + f->open_array_section("quorum_names"); + for (list<string>::iterator p = quorum_names.begin(); p != quorum_names.end(); ++p) + f->dump_string("mon", *p); + f->close_section(); // quorum_names + + f->dump_string("quorum_leader_name", quorum.empty() ? string() : monmap->get_name(*quorum.begin())); + + if (!quorum.empty()) { + f->dump_int( + "quorum_age", + std::chrono::duration_cast<std::chrono::seconds>( + mono_clock::now() - quorum_since).count()); + } + + f->open_object_section("monmap"); + monmap->dump(f); + f->close_section(); // monmap + + f->close_section(); // quorum_status + f->flush(ss); + if (free_formatter) + delete f; +} + +void Monitor::get_mon_status(Formatter *f, ostream& ss) +{ + bool free_formatter = false; + + if (!f) { + // louzy/lazy hack: default to json if no formatter has been defined + f = new JSONFormatter(); + free_formatter = true; + } + + f->open_object_section("mon_status"); + f->dump_string("name", name); + f->dump_int("rank", rank); + f->dump_string("state", get_state_name()); + f->dump_int("election_epoch", get_epoch()); + + f->open_array_section("quorum"); + for (set<int>::iterator p = quorum.begin(); p != quorum.end(); ++p) { + f->dump_int("mon", *p); + } + f->close_section(); // quorum + + if (!quorum.empty()) { + f->dump_int( + "quorum_age", + std::chrono::duration_cast<std::chrono::seconds>( + mono_clock::now() - quorum_since).count()); + } + + f->open_object_section("features"); + f->dump_stream("required_con") << required_features; + mon_feature_t req_mon_features = get_required_mon_features(); + req_mon_features.dump(f, "required_mon"); + f->dump_stream("quorum_con") << quorum_con_features; + quorum_mon_features.dump(f, "quorum_mon"); + f->close_section(); // features + + f->open_array_section("outside_quorum"); + for (set<string>::iterator p = outside_quorum.begin(); p != outside_quorum.end(); ++p) + f->dump_string("mon", *p); + f->close_section(); // outside_quorum + + f->open_array_section("extra_probe_peers"); + for (set<entity_addrvec_t>::iterator p = extra_probe_peers.begin(); + p != extra_probe_peers.end(); + ++p) { + f->dump_object("peer", *p); + } + f->close_section(); // extra_probe_peers + + f->open_array_section("sync_provider"); + for (map<uint64_t,SyncProvider>::const_iterator p = sync_providers.begin(); + p != sync_providers.end(); + ++p) { + f->dump_unsigned("cookie", p->second.cookie); + f->dump_object("addrs", p->second.addrs); + f->dump_stream("timeout") << p->second.timeout; + f->dump_unsigned("last_committed", p->second.last_committed); + f->dump_stream("last_key") << p->second.last_key; + } + f->close_section(); + + if (is_synchronizing()) { + f->open_object_section("sync"); + f->dump_stream("sync_provider") << sync_provider; + f->dump_unsigned("sync_cookie", sync_cookie); + f->dump_unsigned("sync_start_version", sync_start_version); + f->close_section(); + } + + if (g_conf()->mon_sync_provider_kill_at > 0) + f->dump_int("provider_kill_at", g_conf()->mon_sync_provider_kill_at); + if (g_conf()->mon_sync_requester_kill_at > 0) + f->dump_int("requester_kill_at", g_conf()->mon_sync_requester_kill_at); + + f->open_object_section("monmap"); + monmap->dump(f); + f->close_section(); + + f->dump_object("feature_map", session_map.feature_map); + f->close_section(); // mon_status + + if (free_formatter) { + // flush formatter to ss and delete it iff we created the formatter + f->flush(ss); + delete f; + } +} + + +// health status to clog + +void Monitor::health_tick_start() +{ + if (!cct->_conf->mon_health_to_clog || + cct->_conf->mon_health_to_clog_tick_interval <= 0) + return; + + dout(15) << __func__ << dendl; + + health_tick_stop(); + health_tick_event = timer.add_event_after( + cct->_conf->mon_health_to_clog_tick_interval, + new C_MonContext(this, [this](int r) { + if (r < 0) + return; + health_tick_start(); + })); +} + +void Monitor::health_tick_stop() +{ + dout(15) << __func__ << dendl; + + if (health_tick_event) { + timer.cancel_event(health_tick_event); + health_tick_event = NULL; + } +} + +utime_t Monitor::health_interval_calc_next_update() +{ + utime_t now = ceph_clock_now(); + + time_t secs = now.sec(); + int remainder = secs % cct->_conf->mon_health_to_clog_interval; + int adjustment = cct->_conf->mon_health_to_clog_interval - remainder; + utime_t next = utime_t(secs + adjustment, 0); + + dout(20) << __func__ + << " now: " << now << "," + << " next: " << next << "," + << " interval: " << cct->_conf->mon_health_to_clog_interval + << dendl; + + return next; +} + +void Monitor::health_interval_start() +{ + dout(15) << __func__ << dendl; + + if (!cct->_conf->mon_health_to_clog || + cct->_conf->mon_health_to_clog_interval <= 0) { + return; + } + + health_interval_stop(); + utime_t next = health_interval_calc_next_update(); + health_interval_event = new C_MonContext(this, [this](int r) { + if (r < 0) + return; + do_health_to_clog_interval(); + }); + if (!timer.add_event_at(next, health_interval_event)) { + health_interval_event = nullptr; + } +} + +void Monitor::health_interval_stop() +{ + dout(15) << __func__ << dendl; + if (health_interval_event) { + timer.cancel_event(health_interval_event); + } + health_interval_event = NULL; +} + +void Monitor::health_events_cleanup() +{ + health_tick_stop(); + health_interval_stop(); + health_status_cache.reset(); +} + +void Monitor::health_to_clog_update_conf(const std::set<std::string> &changed) +{ + dout(20) << __func__ << dendl; + + if (changed.count("mon_health_to_clog")) { + if (!cct->_conf->mon_health_to_clog) { + health_events_cleanup(); + return; + } else { + if (!health_tick_event) { + health_tick_start(); + } + if (!health_interval_event) { + health_interval_start(); + } + } + } + + if (changed.count("mon_health_to_clog_interval")) { + if (cct->_conf->mon_health_to_clog_interval <= 0) { + health_interval_stop(); + } else { + health_interval_start(); + } + } + + if (changed.count("mon_health_to_clog_tick_interval")) { + if (cct->_conf->mon_health_to_clog_tick_interval <= 0) { + health_tick_stop(); + } else { + health_tick_start(); + } + } +} + +void Monitor::do_health_to_clog_interval() +{ + // outputting to clog may have been disabled in the conf + // since we were scheduled. + if (!cct->_conf->mon_health_to_clog || + cct->_conf->mon_health_to_clog_interval <= 0) + return; + + dout(10) << __func__ << dendl; + + // do we have a cached value for next_clog_update? if not, + // do we know when the last update was? + + do_health_to_clog(true); + health_interval_start(); +} + +void Monitor::do_health_to_clog(bool force) +{ + // outputting to clog may have been disabled in the conf + // since we were scheduled. + if (!cct->_conf->mon_health_to_clog || + cct->_conf->mon_health_to_clog_interval <= 0) + return; + + dout(10) << __func__ << (force ? " (force)" : "") << dendl; + + string summary; + health_status_t level = get_health_status(false, nullptr, &summary); + if (!force && + summary == health_status_cache.summary && + level == health_status_cache.overall) + return; + + if (g_conf()->mon_health_detail_to_clog && + summary != health_status_cache.summary && + level != HEALTH_OK) { + string details; + level = get_health_status(true, nullptr, &details); + clog->health(level) << "Health detail: " << details; + } else { + clog->health(level) << "overall " << summary; + } + health_status_cache.summary = summary; + health_status_cache.overall = level; +} + +health_status_t Monitor::get_health_status( + bool want_detail, + Formatter *f, + std::string *plain, + const char *sep1, + const char *sep2) +{ + health_status_t r = HEALTH_OK; + if (f) { + f->open_object_section("health"); + f->open_object_section("checks"); + } + + string summary; + string *psummary = f ? nullptr : &summary; + for (auto& svc : paxos_service) { + r = std::min(r, svc->get_health_checks().dump_summary( + f, psummary, sep2, want_detail)); + } + + if (f) { + f->close_section(); + f->dump_stream("status") << r; + f->close_section(); + } else { + // one-liner: HEALTH_FOO[ thing1[; thing2 ...]] + *plain = stringify(r); + if (summary.size()) { + *plain += sep1; + *plain += summary; + } + *plain += "\n"; + } + + if (want_detail && !f) { + for (auto& svc : paxos_service) { + svc->get_health_checks().dump_detail(plain); + } + } + + return r; +} + +void Monitor::log_health( + const health_check_map_t& updated, + const health_check_map_t& previous, + MonitorDBStore::TransactionRef t) +{ + if (!g_conf()->mon_health_to_clog) { + return; + } + + const utime_t now = ceph_clock_now(); + + // FIXME: log atomically as part of @t instead of using clog. + dout(10) << __func__ << " updated " << updated.checks.size() + << " previous " << previous.checks.size() + << dendl; + const auto min_log_period = g_conf().get_val<int64_t>( + "mon_health_log_update_period"); + for (auto& p : updated.checks) { + auto q = previous.checks.find(p.first); + bool logged = false; + if (q == previous.checks.end()) { + // new + ostringstream ss; + ss << "Health check failed: " << p.second.summary << " (" + << p.first << ")"; + clog->health(p.second.severity) << ss.str(); + + logged = true; + } else { + if (p.second.summary != q->second.summary || + p.second.severity != q->second.severity) { + + auto status_iter = health_check_log_times.find(p.first); + if (status_iter != health_check_log_times.end()) { + if (p.second.severity == q->second.severity && + now - status_iter->second.updated_at < min_log_period) { + // We already logged this recently and the severity is unchanged, + // so skip emitting an update of the summary string. + // We'll get an update out of tick() later if the check + // is still failing. + continue; + } + } + + // summary or severity changed (ignore detail changes at this level) + ostringstream ss; + ss << "Health check update: " << p.second.summary << " (" << p.first << ")"; + clog->health(p.second.severity) << ss.str(); + + logged = true; + } + } + // Record the time at which we last logged, so that we can check this + // when considering whether/when to print update messages. + if (logged) { + auto iter = health_check_log_times.find(p.first); + if (iter == health_check_log_times.end()) { + health_check_log_times.emplace(p.first, HealthCheckLogStatus( + p.second.severity, p.second.summary, now)); + } else { + iter->second = HealthCheckLogStatus( + p.second.severity, p.second.summary, now); + } + } + } + for (auto& p : previous.checks) { + if (!updated.checks.count(p.first)) { + // cleared + ostringstream ss; + if (p.first == "DEGRADED_OBJECTS") { + clog->info() << "All degraded objects recovered"; + } else if (p.first == "OSD_FLAGS") { + clog->info() << "OSD flags cleared"; + } else { + clog->info() << "Health check cleared: " << p.first << " (was: " + << p.second.summary << ")"; + } + + if (health_check_log_times.count(p.first)) { + health_check_log_times.erase(p.first); + } + } + } + + if (previous.checks.size() && updated.checks.size() == 0) { + // We might be going into a fully healthy state, check + // other subsystems + bool any_checks = false; + for (auto& svc : paxos_service) { + if (&(svc->get_health_checks()) == &(previous)) { + // Ignore the ones we're clearing right now + continue; + } + + if (svc->get_health_checks().checks.size() > 0) { + any_checks = true; + break; + } + } + if (!any_checks) { + clog->info() << "Cluster is now healthy"; + } + } +} + +void Monitor::get_cluster_status(stringstream &ss, Formatter *f) +{ + if (f) + f->open_object_section("status"); + + mono_clock::time_point now = mono_clock::now(); + if (f) { + f->dump_stream("fsid") << monmap->get_fsid(); + get_health_status(false, f, nullptr); + f->dump_unsigned("election_epoch", get_epoch()); + { + f->open_array_section("quorum"); + for (set<int>::iterator p = quorum.begin(); p != quorum.end(); ++p) + f->dump_int("rank", *p); + f->close_section(); + f->open_array_section("quorum_names"); + for (set<int>::iterator p = quorum.begin(); p != quorum.end(); ++p) + f->dump_string("id", monmap->get_name(*p)); + f->close_section(); + f->dump_int( + "quorum_age", + std::chrono::duration_cast<std::chrono::seconds>( + mono_clock::now() - quorum_since).count()); + } + f->open_object_section("monmap"); + monmap->dump_summary(f); + f->close_section(); + f->open_object_section("osdmap"); + osdmon()->osdmap.print_summary(f, cout, string(12, ' ')); + f->close_section(); + f->open_object_section("pgmap"); + mgrstatmon()->print_summary(f, NULL); + f->close_section(); + f->open_object_section("fsmap"); + mdsmon()->get_fsmap().print_summary(f, NULL); + f->close_section(); + f->open_object_section("mgrmap"); + mgrmon()->get_map().print_summary(f, nullptr); + f->close_section(); + + f->dump_object("servicemap", mgrstatmon()->get_service_map()); + + f->open_object_section("progress_events"); + for (auto& i : mgrstatmon()->get_progress_events()) { + f->dump_object(i.first.c_str(), i.second); + } + f->close_section(); + + f->close_section(); + } else { + ss << " cluster:\n"; + ss << " id: " << monmap->get_fsid() << "\n"; + + string health; + get_health_status(false, nullptr, &health, + "\n ", "\n "); + ss << " health: " << health << "\n"; + + ss << "\n \n services:\n"; + { + size_t maxlen = 3; + auto& service_map = mgrstatmon()->get_service_map(); + for (auto& p : service_map.services) { + maxlen = std::max(maxlen, p.first.size()); + } + string spacing(maxlen - 3, ' '); + const auto quorum_names = get_quorum_names(); + const auto mon_count = monmap->mon_info.size(); + ss << " mon: " << spacing << mon_count << " daemons, quorum " + << quorum_names << " (age " << timespan_str(now - quorum_since) << ")"; + if (quorum_names.size() != mon_count) { + std::list<std::string> out_of_q; + for (size_t i = 0; i < monmap->ranks.size(); ++i) { + if (quorum.count(i) == 0) { + out_of_q.push_back(monmap->ranks[i]); + } + } + ss << ", out of quorum: " << joinify(out_of_q.begin(), + out_of_q.end(), std::string(", ")); + } + ss << "\n"; + if (mgrmon()->in_use()) { + ss << " mgr: " << spacing; + mgrmon()->get_map().print_summary(nullptr, &ss); + ss << "\n"; + } + if (mdsmon()->should_print_status()) { + ss << " mds: " << spacing << mdsmon()->get_fsmap() << "\n"; + } + ss << " osd: " << spacing; + osdmon()->osdmap.print_summary(NULL, ss, string(maxlen + 6, ' ')); + ss << "\n"; + for (auto& p : service_map.services) { + const std::string &service = p.first; + // filter out normal ceph entity types + if (ServiceMap::is_normal_ceph_entity(service)) { + continue; + } + ss << " " << p.first << ": " << string(maxlen - p.first.size(), ' ') + << p.second.get_summary() << "\n"; + } + } + + { + auto& service_map = mgrstatmon()->get_service_map(); + if (!service_map.services.empty()) { + ss << "\n \n task status:\n"; + { + for (auto &p : service_map.services) { + ss << p.second.get_task_summary(p.first); + } + } + } + } + + ss << "\n \n data:\n"; + mgrstatmon()->print_summary(NULL, &ss); + + auto& pem = mgrstatmon()->get_progress_events(); + if (!pem.empty()) { + ss << "\n \n progress:\n"; + for (auto& i : pem) { + ss << " " << i.second.message << "\n"; + ss << " ["; + unsigned j; + for (j = 0; j < (unsigned)(i.second.progress * 30.0); ++j) { + ss << '='; + } + for (; j < 30; ++j) { + ss << '.'; + } + ss << "]\n"; + } + } + ss << "\n "; + } +} + +void Monitor::_generate_command_map(cmdmap_t& cmdmap, + map<string,string> ¶m_str_map) +{ + for (auto p = cmdmap.begin(); p != cmdmap.end(); ++p) { + if (p->first == "prefix") + continue; + if (p->first == "caps") { + vector<string> cv; + if (cmd_getval(g_ceph_context, cmdmap, "caps", cv) && + cv.size() % 2 == 0) { + for (unsigned i = 0; i < cv.size(); i += 2) { + string k = string("caps_") + cv[i]; + param_str_map[k] = cv[i + 1]; + } + continue; + } + } + param_str_map[p->first] = cmd_vartype_stringify(p->second); + } +} + +const MonCommand *Monitor::_get_moncommand( + const string &cmd_prefix, + const vector<MonCommand>& cmds) +{ + for (auto& c : cmds) { + if (c.cmdstring.compare(0, cmd_prefix.size(), cmd_prefix) == 0) { + return &c; + } + } + return nullptr; +} + +bool Monitor::_allowed_command(MonSession *s, const string &module, + const string &prefix, const cmdmap_t& cmdmap, + const map<string,string>& param_str_map, + const MonCommand *this_cmd) { + + bool cmd_r = this_cmd->requires_perm('r'); + bool cmd_w = this_cmd->requires_perm('w'); + bool cmd_x = this_cmd->requires_perm('x'); + + bool capable = s->caps.is_capable( + g_ceph_context, + s->entity_name, + module, prefix, param_str_map, + cmd_r, cmd_w, cmd_x, + s->get_peer_socket_addr()); + + dout(10) << __func__ << " " << (capable ? "" : "not ") << "capable" << dendl; + return capable; +} + +void Monitor::format_command_descriptions(const std::vector<MonCommand> &commands, + Formatter *f, + uint64_t features, + bufferlist *rdata) +{ + int cmdnum = 0; + f->open_object_section("command_descriptions"); + for (const auto &cmd : commands) { + unsigned flags = cmd.flags; + ostringstream secname; + secname << "cmd" << setfill('0') << std::setw(3) << cmdnum; + dump_cmddesc_to_json(f, features, secname.str(), + cmd.cmdstring, cmd.helpstring, cmd.module, + cmd.req_perms, flags); + cmdnum++; + } + f->close_section(); // command_descriptions + + f->flush(*rdata); +} + +bool Monitor::is_keyring_required() +{ + return auth_cluster_required.is_supported_auth(CEPH_AUTH_CEPHX) || + auth_service_required.is_supported_auth(CEPH_AUTH_CEPHX) || + auth_cluster_required.is_supported_auth(CEPH_AUTH_GSS) || + auth_service_required.is_supported_auth(CEPH_AUTH_GSS); +} + +struct C_MgrProxyCommand : public Context { + Monitor *mon; + MonOpRequestRef op; + uint64_t size; + bufferlist outbl; + string outs; + C_MgrProxyCommand(Monitor *mon, MonOpRequestRef op, uint64_t s) + : mon(mon), op(op), size(s) { } + void finish(int r) { + std::lock_guard l(mon->lock); + mon->mgr_proxy_bytes -= size; + mon->reply_command(op, r, outs, outbl, 0); + } +}; + +void Monitor::handle_command(MonOpRequestRef op) +{ + ceph_assert(op->is_type_command()); + MMonCommand *m = static_cast<MMonCommand*>(op->get_req()); + if (m->fsid != monmap->fsid) { + dout(0) << "handle_command on fsid " << m->fsid << " != " << monmap->fsid << dendl; + reply_command(op, -EPERM, "wrong fsid", 0); + return; + } + + MonSession *session = op->get_session(); + if (!session) { + dout(5) << __func__ << " dropping stray message " << *m << dendl; + return; + } + + if (m->cmd.empty()) { + string rs = "No command supplied"; + reply_command(op, -EINVAL, rs, 0); + return; + } + + string prefix; + vector<string> fullcmd; + cmdmap_t cmdmap; + stringstream ss, ds; + bufferlist rdata; + string rs; + int r = -EINVAL; + rs = "unrecognized command"; + + if (!cmdmap_from_json(m->cmd, &cmdmap, ss)) { + // ss has reason for failure + r = -EINVAL; + rs = ss.str(); + if (!m->get_source().is_mon()) // don't reply to mon->mon commands + reply_command(op, r, rs, 0); + return; + } + + // check return value. If no prefix parameter provided, + // return value will be false, then return error info. + if (!cmd_getval(g_ceph_context, cmdmap, "prefix", prefix)) { + reply_command(op, -EINVAL, "command prefix not found", 0); + return; + } + + // check prefix is empty + if (prefix.empty()) { + reply_command(op, -EINVAL, "command prefix must not be empty", 0); + return; + } + + if (prefix == "get_command_descriptions") { + bufferlist rdata; + Formatter *f = Formatter::create("json"); + + std::vector<MonCommand> commands = static_cast<MgrMonitor*>( + paxos_service[PAXOS_MGR].get())->get_command_descs(); + + for (auto& c : leader_mon_commands) { + commands.push_back(c); + } + + auto features = m->get_connection()->get_features(); + format_command_descriptions(commands, f, features, &rdata); + delete f; + reply_command(op, 0, "", rdata, 0); + return; + } + + string module; + string err; + + dout(0) << "handle_command " << *m << dendl; + + string format; + cmd_getval(g_ceph_context, cmdmap, "format", format, string("plain")); + boost::scoped_ptr<Formatter> f(Formatter::create(format)); + + get_str_vec(prefix, fullcmd); + + // make sure fullcmd is not empty. + // invalid prefix will cause empty vector fullcmd. + // such as, prefix=";,,;" + if (fullcmd.empty()) { + reply_command(op, -EINVAL, "command requires a prefix to be valid", 0); + return; + } + + module = fullcmd[0]; + + // validate command is in leader map + + const MonCommand *leader_cmd; + const auto& mgr_cmds = mgrmon()->get_command_descs(); + const MonCommand *mgr_cmd = nullptr; + if (!mgr_cmds.empty()) { + mgr_cmd = _get_moncommand(prefix, mgr_cmds); + } + leader_cmd = _get_moncommand(prefix, leader_mon_commands); + if (!leader_cmd) { + leader_cmd = mgr_cmd; + if (!leader_cmd) { + reply_command(op, -EINVAL, "command not known", 0); + return; + } + } + // validate command is in our map & matches, or forward if it is allowed + const MonCommand *mon_cmd = _get_moncommand( + prefix, + get_local_commands(quorum_mon_features)); + if (!mon_cmd) { + mon_cmd = mgr_cmd; + } + if (!is_leader()) { + if (!mon_cmd) { + if (leader_cmd->is_noforward()) { + reply_command(op, -EINVAL, + "command not locally supported and not allowed to forward", + 0); + return; + } + dout(10) << "Command not locally supported, forwarding request " + << m << dendl; + forward_request_leader(op); + return; + } else if (!mon_cmd->is_compat(leader_cmd)) { + if (mon_cmd->is_noforward()) { + reply_command(op, -EINVAL, + "command not compatible with leader and not allowed to forward", + 0); + return; + } + dout(10) << "Command not compatible with leader, forwarding request " + << m << dendl; + forward_request_leader(op); + return; + } + } + + if (mon_cmd->is_obsolete() || + (cct->_conf->mon_debug_deprecated_as_obsolete + && mon_cmd->is_deprecated())) { + reply_command(op, -ENOTSUP, + "command is obsolete; please check usage and/or man page", + 0); + return; + } + + if (session->proxy_con && mon_cmd->is_noforward()) { + dout(10) << "Got forward for noforward command " << m << dendl; + reply_command(op, -EINVAL, "forward for noforward command", rdata, 0); + return; + } + + /* what we perceive as being the service the command falls under */ + string service(mon_cmd->module); + + dout(25) << __func__ << " prefix='" << prefix + << "' module='" << module + << "' service='" << service << "'" << dendl; + + bool cmd_is_rw = + (mon_cmd->requires_perm('w') || mon_cmd->requires_perm('x')); + + // validate user's permissions for requested command + map<string,string> param_str_map; + _generate_command_map(cmdmap, param_str_map); + if (!_allowed_command(session, service, prefix, cmdmap, + param_str_map, mon_cmd)) { + dout(1) << __func__ << " access denied" << dendl; + if (prefix != "config set" && prefix != "config-key set") + (cmd_is_rw ? audit_clog->info() : audit_clog->debug()) + << "from='" << session->name << " " << session->addrs << "' " + << "entity='" << session->entity_name << "' " + << "cmd=" << m->cmd << ": access denied"; + reply_command(op, -EACCES, "access denied", 0); + return; + } + + if (prefix != "config set" && prefix != "config-key set") + (cmd_is_rw ? audit_clog->info() : audit_clog->debug()) + << "from='" << session->name << " " << session->addrs << "' " + << "entity='" << session->entity_name << "' " + << "cmd=" << m->cmd << ": dispatch"; + + if (mon_cmd->is_mgr()) { + const auto& hdr = m->get_header(); + uint64_t size = hdr.front_len + hdr.middle_len + hdr.data_len; + uint64_t max = g_conf().get_val<Option::size_t>("mon_client_bytes") + * g_conf().get_val<double>("mon_mgr_proxy_client_bytes_ratio"); + if (mgr_proxy_bytes + size > max) { + dout(10) << __func__ << " current mgr proxy bytes " << mgr_proxy_bytes + << " + " << size << " > max " << max << dendl; + reply_command(op, -EAGAIN, "hit limit on proxied mgr commands", rdata, 0); + return; + } + mgr_proxy_bytes += size; + dout(10) << __func__ << " proxying mgr command (+" << size + << " -> " << mgr_proxy_bytes << ")" << dendl; + C_MgrProxyCommand *fin = new C_MgrProxyCommand(this, op, size); + mgr_client.start_command(m->cmd, + m->get_data(), + &fin->outbl, + &fin->outs, + new C_OnFinisher(fin, &finisher)); + return; + } + + if ((module == "mds" || module == "fs") && + prefix != "fs authorize") { + mdsmon()->dispatch(op); + return; + } + if ((module == "osd" || + prefix == "pg map" || + prefix == "pg repeer") && + prefix != "osd last-stat-seq") { + osdmon()->dispatch(op); + return; + } + if (module == "config") { + configmon()->dispatch(op); + return; + } + + if (module == "mon" && + /* Let the Monitor class handle the following commands: + * 'mon compact' + * 'mon scrub' + * 'mon sync force' + */ + prefix != "mon compact" && + prefix != "mon scrub" && + prefix != "mon sync force" && + prefix != "mon metadata" && + prefix != "mon versions" && + prefix != "mon count-metadata" && + prefix != "mon ok-to-stop" && + prefix != "mon ok-to-add-offline" && + prefix != "mon ok-to-rm") { + monmon()->dispatch(op); + return; + } + if (module == "auth" || prefix == "fs authorize") { + authmon()->dispatch(op); + return; + } + if (module == "log") { + logmon()->dispatch(op); + return; + } + + if (module == "config-key") { + config_key_service->dispatch(op); + return; + } + + if (module == "mgr") { + mgrmon()->dispatch(op); + return; + } + + if (prefix == "fsid") { + if (f) { + f->open_object_section("fsid"); + f->dump_stream("fsid") << monmap->fsid; + f->close_section(); + f->flush(rdata); + } else { + ds << monmap->fsid; + rdata.append(ds); + } + reply_command(op, 0, "", rdata, 0); + return; + } + + if (prefix == "scrub" || prefix == "mon scrub") { + wait_for_paxos_write(); + if (is_leader()) { + int r = scrub_start(); + reply_command(op, r, "", rdata, 0); + } else if (is_peon()) { + forward_request_leader(op); + } else { + reply_command(op, -EAGAIN, "no quorum", rdata, 0); + } + return; + } + + if (prefix == "compact" || prefix == "mon compact") { + dout(1) << "triggering manual compaction" << dendl; + auto start = ceph::coarse_mono_clock::now(); + store->compact_async(); + auto end = ceph::coarse_mono_clock::now(); + double duration = std::chrono::duration<double>(end-start).count(); + dout(1) << "finished manual compaction in " << duration << " seconds" << dendl; + ostringstream oss; + oss << "compacted " << g_conf().get_val<std::string>("mon_keyvaluedb") << " in " << duration << " seconds"; + rs = oss.str(); + r = 0; + } + else if (prefix == "injectargs") { + vector<string> injected_args; + cmd_getval(g_ceph_context, cmdmap, "injected_args", injected_args); + if (!injected_args.empty()) { + dout(0) << "parsing injected options '" << injected_args << "'" << dendl; + ostringstream oss; + r = g_conf().injectargs(str_join(injected_args, " "), &oss); + ss << "injectargs:" << oss.str(); + rs = ss.str(); + goto out; + } else { + rs = "must supply options to be parsed in a single string"; + r = -EINVAL; + } + } else if (prefix == "time-sync-status") { + if (!f) + f.reset(Formatter::create("json-pretty")); + f->open_object_section("time_sync"); + if (!timecheck_skews.empty()) { + f->open_object_section("time_skew_status"); + for (auto& i : timecheck_skews) { + double skew = i.second; + double latency = timecheck_latencies[i.first]; + string name = monmap->get_name(i.first); + ostringstream tcss; + health_status_t tcstatus = timecheck_status(tcss, skew, latency); + f->open_object_section(name.c_str()); + f->dump_float("skew", skew); + f->dump_float("latency", latency); + f->dump_stream("health") << tcstatus; + if (tcstatus != HEALTH_OK) { + f->dump_stream("details") << tcss.str(); + } + f->close_section(); + } + f->close_section(); + } + f->open_object_section("timechecks"); + f->dump_unsigned("epoch", get_epoch()); + f->dump_int("round", timecheck_round); + f->dump_stream("round_status") << ((timecheck_round%2) ? + "on-going" : "finished"); + f->close_section(); + f->close_section(); + f->flush(rdata); + r = 0; + rs = ""; + } else if (prefix == "config set") { + std::string key; + cmd_getval(cct, cmdmap, "key", key); + std::string val; + cmd_getval(cct, cmdmap, "value", val); + r = g_conf().set_val(key, val, &ss); + if (r == 0) { + g_conf().apply_changes(nullptr); + } + rs = ss.str(); + goto out; + } else if (prefix == "status" || + prefix == "health" || + prefix == "df") { + string detail; + cmd_getval(g_ceph_context, cmdmap, "detail", detail); + + if (prefix == "status") { + // get_cluster_status handles f == NULL + get_cluster_status(ds, f.get()); + + if (f) { + f->flush(ds); + ds << '\n'; + } + rdata.append(ds); + } else if (prefix == "health") { + string plain; + get_health_status(detail == "detail", f.get(), f ? nullptr : &plain); + if (f) { + f->flush(rdata); + } else { + rdata.append(plain); + } + } else if (prefix == "df") { + bool verbose = (detail == "detail"); + if (f) + f->open_object_section("stats"); + + mgrstatmon()->dump_cluster_stats(&ds, f.get(), verbose); + if (!f) { + ds << "\n \n"; + } + mgrstatmon()->dump_pool_stats(osdmon()->osdmap, &ds, f.get(), verbose); + + if (f) { + f->close_section(); + f->flush(ds); + ds << '\n'; + } + } else { + ceph_abort_msg("We should never get here!"); + return; + } + rdata.append(ds); + rs = ""; + r = 0; + } else if (prefix == "report") { + + // this must be formatted, in its current form + if (!f) + f.reset(Formatter::create("json-pretty")); + f->open_object_section("report"); + f->dump_stream("cluster_fingerprint") << fingerprint; + f->dump_string("version", ceph_version_to_str()); + f->dump_string("commit", git_version_to_str()); + f->dump_stream("timestamp") << ceph_clock_now(); + + vector<string> tagsvec; + cmd_getval(g_ceph_context, cmdmap, "tags", tagsvec); + string tagstr = str_join(tagsvec, " "); + if (!tagstr.empty()) + tagstr = tagstr.substr(0, tagstr.find_last_of(' ')); + f->dump_string("tag", tagstr); + + get_health_status(true, f.get(), nullptr); + + monmon()->dump_info(f.get()); + osdmon()->dump_info(f.get()); + mdsmon()->dump_info(f.get()); + authmon()->dump_info(f.get()); + mgrstatmon()->dump_info(f.get()); + + paxos->dump_info(f.get()); + + f->close_section(); + f->flush(rdata); + + ostringstream ss2; + ss2 << "report " << rdata.crc32c(CEPH_MON_PORT_LEGACY); + rs = ss2.str(); + r = 0; + } else if (prefix == "osd last-stat-seq") { + int64_t osd; + cmd_getval(g_ceph_context, cmdmap, "id", osd); + uint64_t seq = mgrstatmon()->get_last_osd_stat_seq(osd); + if (f) { + f->dump_unsigned("seq", seq); + f->flush(ds); + } else { + ds << seq; + rdata.append(ds); + } + rs = ""; + r = 0; + } else if (prefix == "node ls") { + string node_type("all"); + cmd_getval(g_ceph_context, cmdmap, "type", node_type); + if (!f) + f.reset(Formatter::create("json-pretty")); + if (node_type == "all") { + f->open_object_section("nodes"); + print_nodes(f.get(), ds); + osdmon()->print_nodes(f.get()); + mdsmon()->print_nodes(f.get()); + mgrmon()->print_nodes(f.get()); + f->close_section(); + } else if (node_type == "mon") { + print_nodes(f.get(), ds); + } else if (node_type == "osd") { + osdmon()->print_nodes(f.get()); + } else if (node_type == "mds") { + mdsmon()->print_nodes(f.get()); + } else if (node_type == "mgr") { + mgrmon()->print_nodes(f.get()); + } + f->flush(ds); + rdata.append(ds); + rs = ""; + r = 0; + } else if (prefix == "features") { + if (!is_leader() && !is_peon()) { + dout(10) << " waiting for quorum" << dendl; + waitfor_quorum.push_back(new C_RetryMessage(this, op)); + return; + } + if (!is_leader()) { + forward_request_leader(op); + return; + } + if (!f) + f.reset(Formatter::create("json-pretty")); + FeatureMap fm; + get_combined_feature_map(&fm); + f->dump_object("features", fm); + f->flush(rdata); + rs = ""; + r = 0; + } else if (prefix == "mon metadata") { + if (!f) + f.reset(Formatter::create("json-pretty")); + + string name; + bool all = !cmd_getval(g_ceph_context, cmdmap, "id", name); + if (!all) { + // Dump a single mon's metadata + int mon = monmap->get_rank(name); + if (mon < 0) { + rs = "requested mon not found"; + r = -ENOENT; + goto out; + } + f->open_object_section("mon_metadata"); + r = get_mon_metadata(mon, f.get(), ds); + f->close_section(); + } else { + // Dump all mons' metadata + r = 0; + f->open_array_section("mon_metadata"); + for (unsigned int rank = 0; rank < monmap->size(); ++rank) { + std::ostringstream get_err; + f->open_object_section("mon"); + f->dump_string("name", monmap->get_name(rank)); + r = get_mon_metadata(rank, f.get(), get_err); + f->close_section(); + if (r == -ENOENT || r == -EINVAL) { + dout(1) << get_err.str() << dendl; + // Drop error, list what metadata we do have + r = 0; + } else if (r != 0) { + derr << "Unexpected error from get_mon_metadata: " + << cpp_strerror(r) << dendl; + ds << get_err.str(); + break; + } + } + f->close_section(); + } + + f->flush(ds); + rdata.append(ds); + rs = ""; + } else if (prefix == "mon versions") { + if (!f) + f.reset(Formatter::create("json-pretty")); + count_metadata("ceph_version", f.get()); + f->flush(ds); + rdata.append(ds); + rs = ""; + r = 0; + } else if (prefix == "mon count-metadata") { + if (!f) + f.reset(Formatter::create("json-pretty")); + string field; + cmd_getval(g_ceph_context, cmdmap, "property", field); + count_metadata(field, f.get()); + f->flush(ds); + rdata.append(ds); + rs = ""; + r = 0; + } else if (prefix == "quorum_status") { + // make sure our map is readable and up to date + if (!is_leader() && !is_peon()) { + dout(10) << " waiting for quorum" << dendl; + waitfor_quorum.push_back(new C_RetryMessage(this, op)); + return; + } + _quorum_status(f.get(), ds); + rdata.append(ds); + rs = ""; + r = 0; + } else if (prefix == "mon ok-to-stop") { + vector<string> ids; + if (!cmd_getval(g_ceph_context, cmdmap, "ids", ids)) { + r = -EINVAL; + goto out; + } + set<string> wouldbe; + for (auto rank : quorum) { + wouldbe.insert(monmap->get_name(rank)); + } + for (auto& n : ids) { + if (monmap->contains(n)) { + wouldbe.erase(n); + } + } + if (wouldbe.size() < monmap->min_quorum_size()) { + r = -EBUSY; + rs = "not enough monitors would be available (" + stringify(wouldbe) + + ") after stopping mons " + stringify(ids); + goto out; + } + r = 0; + rs = "quorum should be preserved (" + stringify(wouldbe) + + ") after stopping " + stringify(ids); + } else if (prefix == "mon ok-to-add-offline") { + if (quorum.size() < monmap->min_quorum_size(monmap->size() + 1)) { + rs = "adding a monitor may break quorum (until that monitor starts)"; + r = -EBUSY; + goto out; + } + rs = "adding another mon that is not yet online will not break quorum"; + r = 0; + } else if (prefix == "mon ok-to-rm") { + string id; + if (!cmd_getval(g_ceph_context, cmdmap, "id", id)) { + r = -EINVAL; + rs = "must specify a monitor id"; + goto out; + } + if (!monmap->contains(id)) { + r = 0; + rs = "mon." + id + " does not exist"; + goto out; + } + int rank = monmap->get_rank(id); + if (quorum.count(rank) && + quorum.size() - 1 < monmap->min_quorum_size(monmap->size() - 1)) { + r = -EBUSY; + rs = "removing mon." + id + " would break quorum"; + goto out; + } + r = 0; + rs = "safe to remove mon." + id; + } else if (prefix == "mon_status") { + get_mon_status(f.get(), ds); + if (f) + f->flush(ds); + rdata.append(ds); + rs = ""; + r = 0; + } else if (prefix == "sync force" || + prefix == "mon sync force") { + bool validate1 = false; + cmd_getval(g_ceph_context, cmdmap, "yes_i_really_mean_it", validate1); + bool validate2 = false; + cmd_getval(g_ceph_context, cmdmap, "i_know_what_i_am_doing", validate2); + + if (!validate1 || !validate2) { + r = -EINVAL; + rs = "are you SURE? this will mean the monitor store will be " + "erased. pass '--yes-i-really-mean-it " + "--i-know-what-i-am-doing' if you really do."; + goto out; + } + sync_force(f.get(), ds); + rs = ds.str(); + r = 0; + } else if (prefix == "heap") { + if (!ceph_using_tcmalloc()) + rs = "tcmalloc not enabled, can't use heap profiler commands\n"; + else { + string heapcmd; + cmd_getval(g_ceph_context, cmdmap, "heapcmd", heapcmd); + // XXX 1-element vector, change at callee or make vector here? + vector<string> heapcmd_vec; + get_str_vec(heapcmd, heapcmd_vec); + string value; + if (cmd_getval(g_ceph_context, cmdmap, "value", value)) + heapcmd_vec.push_back(value); + ceph_heap_profiler_handle_command(heapcmd_vec, ds); + rdata.append(ds); + rs = ""; + r = 0; + } + } else if (prefix == "quorum") { + string quorumcmd; + cmd_getval(g_ceph_context, cmdmap, "quorumcmd", quorumcmd); + if (quorumcmd == "exit") { + start_election(); + elector.stop_participating(); + rs = "stopped responding to quorum, initiated new election"; + r = 0; + } else if (quorumcmd == "enter") { + elector.start_participating(); + start_election(); + rs = "started responding to quorum, initiated new election"; + r = 0; + } else { + rs = "needs a valid 'quorum' command"; + r = -EINVAL; + } + } else if (prefix == "version") { + if (f) { + f->open_object_section("version"); + f->dump_string("version", pretty_version_to_str()); + f->close_section(); + f->flush(ds); + } else { + ds << pretty_version_to_str(); + } + rdata.append(ds); + rs = ""; + r = 0; + } else if (prefix == "versions") { + if (!f) + f.reset(Formatter::create("json-pretty")); + map<string,int> overall; + f->open_object_section("version"); + map<string,int> mon, mgr, osd, mds; + + count_metadata("ceph_version", &mon); + f->open_object_section("mon"); + for (auto& p : mon) { + f->dump_int(p.first.c_str(), p.second); + overall[p.first] += p.second; + } + f->close_section(); + + mgrmon()->count_metadata("ceph_version", &mgr); + f->open_object_section("mgr"); + for (auto& p : mgr) { + f->dump_int(p.first.c_str(), p.second); + overall[p.first] += p.second; + } + f->close_section(); + + osdmon()->count_metadata("ceph_version", &osd); + f->open_object_section("osd"); + for (auto& p : osd) { + f->dump_int(p.first.c_str(), p.second); + overall[p.first] += p.second; + } + f->close_section(); + + mdsmon()->count_metadata("ceph_version", &mds); + f->open_object_section("mds"); + for (auto& p : mds) { + f->dump_int(p.first.c_str(), p.second); + overall[p.first] += p.second; + } + f->close_section(); + + for (auto& p : mgrstatmon()->get_service_map().services) { + auto &service = p.first; + if (ServiceMap::is_normal_ceph_entity(service)) { + continue; + } + f->open_object_section(service.c_str()); + map<string,int> m; + p.second.count_metadata("ceph_version", &m); + for (auto& q : m) { + f->dump_int(q.first.c_str(), q.second); + overall[q.first] += q.second; + } + f->close_section(); + } + + f->open_object_section("overall"); + for (auto& p : overall) { + f->dump_int(p.first.c_str(), p.second); + } + f->close_section(); + f->close_section(); + f->flush(rdata); + rs = ""; + r = 0; + } else if (prefix == "smart") { + string want_devid; + cmd_getval(cct, cmdmap, "devid", want_devid); + + string devname = store->get_devname(); + set<string> devnames; + get_raw_devices(devname, &devnames); + json_spirit::mObject json_map; + uint64_t smart_timeout = cct->_conf.get_val<uint64_t>( + "mon_smart_report_timeout"); + for (auto& devname : devnames) { + string err; + string devid = get_device_id(devname, &err); + if (want_devid.size() && want_devid != devid) { + derr << "get_device_id failed on " << devname << ": " << err << dendl; + continue; + } + json_spirit::mValue smart_json; + if (block_device_get_metrics(devname, smart_timeout, + &smart_json)) { + dout(10) << "block_device_get_metrics failed for /dev/" << devname + << dendl; + continue; + } + json_map[devid] = smart_json; + } + ostringstream ss; + json_spirit::write(json_map, ss, json_spirit::pretty_print); + rdata.append(ss.str()); + r = 0; + rs = ""; + } + + out: + if (!m->get_source().is_mon()) // don't reply to mon->mon commands + reply_command(op, r, rs, rdata, 0); +} + +void Monitor::reply_command(MonOpRequestRef op, int rc, const string &rs, version_t version) +{ + bufferlist rdata; + reply_command(op, rc, rs, rdata, version); +} + +void Monitor::reply_command(MonOpRequestRef op, int rc, const string &rs, + bufferlist& rdata, version_t version) +{ + MMonCommand *m = static_cast<MMonCommand*>(op->get_req()); + ceph_assert(m->get_type() == MSG_MON_COMMAND); + MMonCommandAck *reply = new MMonCommandAck(m->cmd, rc, rs, version); + reply->set_tid(m->get_tid()); + reply->set_data(rdata); + send_reply(op, reply); +} + + +// ------------------------ +// request/reply routing +// +// a client/mds/osd will connect to a random monitor. we need to forward any +// messages requiring state updates to the leader, and then route any replies +// back via the correct monitor and back to them. (the monitor will not +// initiate any connections.) + +void Monitor::forward_request_leader(MonOpRequestRef op) +{ + op->mark_event(__func__); + + int mon = get_leader(); + MonSession *session = op->get_session(); + PaxosServiceMessage *req = op->get_req<PaxosServiceMessage>(); + + if (req->get_source().is_mon() && req->get_source_addrs() != messenger->get_myaddrs()) { + dout(10) << "forward_request won't forward (non-local) mon request " << *req << dendl; + } else if (session->proxy_con) { + dout(10) << "forward_request won't double fwd request " << *req << dendl; + } else if (!session->closed) { + RoutedRequest *rr = new RoutedRequest; + rr->tid = ++routed_request_tid; + rr->con = req->get_connection(); + rr->con_features = rr->con->get_features(); + encode_message(req, CEPH_FEATURES_ALL, rr->request_bl); // for my use only; use all features + rr->session = static_cast<MonSession *>(session->get()); + rr->op = op; + routed_requests[rr->tid] = rr; + session->routed_request_tids.insert(rr->tid); + + dout(10) << "forward_request " << rr->tid << " request " << *req + << " features " << rr->con_features << dendl; + + MForward *forward = new MForward(rr->tid, + req, + rr->con_features, + rr->session->caps); + forward->set_priority(req->get_priority()); + if (session->auth_handler) { + forward->entity_name = session->entity_name; + } else if (req->get_source().is_mon()) { + forward->entity_name.set_type(CEPH_ENTITY_TYPE_MON); + } + send_mon_message(forward, mon); + op->mark_forwarded(); + ceph_assert(op->get_req()->get_type() != 0); + } else { + dout(10) << "forward_request no session for request " << *req << dendl; + } +} + +// fake connection attached to forwarded messages +struct AnonConnection : public Connection { + entity_addr_t socket_addr; + explicit AnonConnection(CephContext *cct, + const entity_addr_t& sa) + : Connection(cct, NULL), + socket_addr(sa) {} + + int send_message(Message *m) override { + ceph_assert(!"send_message on anonymous connection"); + } + void send_keepalive() override { + ceph_assert(!"send_keepalive on anonymous connection"); + } + void mark_down() override { + // silently ignore + } + void mark_disposable() override { + // silengtly ignore + } + bool is_connected() override { return false; } + entity_addr_t get_peer_socket_addr() const override { + return socket_addr; + } +}; + +//extract the original message and put it into the regular dispatch function +void Monitor::handle_forward(MonOpRequestRef op) +{ + MForward *m = static_cast<MForward*>(op->get_req()); + dout(10) << "received forwarded message from " + << ceph_entity_type_name(m->client_type) + << " " << m->client_addrs + << " via " << m->get_source_inst() << dendl; + MonSession *session = op->get_session(); + ceph_assert(session); + + if (!session->is_capable("mon", MON_CAP_X)) { + dout(0) << "forward from entity with insufficient caps! " + << session->caps << dendl; + } else { + // see PaxosService::dispatch(); we rely on this being anon + // (c->msgr == NULL) + PaxosServiceMessage *req = m->claim_message(); + ceph_assert(req != NULL); + + ConnectionRef c(new AnonConnection(cct, m->client_socket_addr)); + MonSession *s = new MonSession(static_cast<Connection*>(c.get())); + s->_ident(req->get_source(), + req->get_source_addrs()); + c->set_priv(RefCountedPtr{s, false}); + c->set_peer_addrs(m->client_addrs); + c->set_peer_type(m->client_type); + c->set_features(m->con_features); + + s->authenticated = true; + s->caps = m->client_caps; + dout(10) << " caps are " << s->caps << dendl; + s->entity_name = m->entity_name; + dout(10) << " entity name '" << s->entity_name << "' type " + << s->entity_name.get_type() << dendl; + s->proxy_con = m->get_connection(); + s->proxy_tid = m->tid; + + req->set_connection(c); + + // not super accurate, but better than nothing. + req->set_recv_stamp(m->get_recv_stamp()); + + /* + * note which election epoch this is; we will drop the message if + * there is a future election since our peers will resend routed + * requests in that case. + */ + req->rx_election_epoch = get_epoch(); + + dout(10) << " mesg " << req << " from " << m->get_source_addr() << dendl; + _ms_dispatch(req); + + // break the session <-> con ref loop by removing the con->session + // reference, which is no longer needed once the MonOpRequest is + // set up. + c->set_priv(NULL); + } +} + +void Monitor::send_reply(MonOpRequestRef op, Message *reply) +{ + op->mark_event(__func__); + + MonSession *session = op->get_session(); + ceph_assert(session); + Message *req = op->get_req(); + ConnectionRef con = op->get_connection(); + + reply->set_cct(g_ceph_context); + dout(2) << __func__ << " " << op << " " << reply << " " << *reply << dendl; + + if (!con) { + dout(2) << "send_reply no connection, dropping reply " << *reply + << " to " << req << " " << *req << dendl; + reply->put(); + op->mark_event("reply: no connection"); + return; + } + + if (!session->con && !session->proxy_con) { + dout(2) << "send_reply no connection, dropping reply " << *reply + << " to " << req << " " << *req << dendl; + reply->put(); + op->mark_event("reply: no connection"); + return; + } + + if (session->proxy_con) { + dout(15) << "send_reply routing reply to " << con->get_peer_addr() + << " via " << session->proxy_con->get_peer_addr() + << " for request " << *req << dendl; + session->proxy_con->send_message(new MRoute(session->proxy_tid, reply)); + op->mark_event("reply: send routed request"); + } else { + session->con->send_message(reply); + op->mark_event("reply: send"); + } +} + +void Monitor::no_reply(MonOpRequestRef op) +{ + MonSession *session = op->get_session(); + Message *req = op->get_req(); + + if (session->proxy_con) { + dout(10) << "no_reply to " << req->get_source_inst() + << " via " << session->proxy_con->get_peer_addr() + << " for request " << *req << dendl; + session->proxy_con->send_message(new MRoute(session->proxy_tid, NULL)); + op->mark_event("no_reply: send routed request"); + } else { + dout(10) << "no_reply to " << req->get_source_inst() + << " " << *req << dendl; + op->mark_event("no_reply"); + } +} + +void Monitor::handle_route(MonOpRequestRef op) +{ + MRoute *m = static_cast<MRoute*>(op->get_req()); + MonSession *session = op->get_session(); + //check privileges + if (!session->is_capable("mon", MON_CAP_X)) { + dout(0) << "MRoute received from entity without appropriate perms! " + << dendl; + return; + } + if (m->msg) + dout(10) << "handle_route tid " << m->session_mon_tid << " " << *m->msg + << dendl; + else + dout(10) << "handle_route tid " << m->session_mon_tid << " null" << dendl; + + // look it up + if (m->session_mon_tid) { + if (routed_requests.count(m->session_mon_tid)) { + RoutedRequest *rr = routed_requests[m->session_mon_tid]; + + // reset payload, in case encoding is dependent on target features + if (m->msg) { + m->msg->clear_payload(); + rr->con->send_message(m->msg); + m->msg = NULL; + } + if (m->send_osdmap_first) { + dout(10) << " sending osdmaps from " << m->send_osdmap_first << dendl; + osdmon()->send_incremental(m->send_osdmap_first, rr->session, + true, MonOpRequestRef()); + } + ceph_assert(rr->tid == m->session_mon_tid && rr->session->routed_request_tids.count(m->session_mon_tid)); + routed_requests.erase(m->session_mon_tid); + rr->session->routed_request_tids.erase(m->session_mon_tid); + delete rr; + } else { + dout(10) << " don't have routed request tid " << m->session_mon_tid << dendl; + } + } else { + dout(10) << " not a routed request, ignoring" << dendl; + } +} + +void Monitor::resend_routed_requests() +{ + dout(10) << "resend_routed_requests" << dendl; + int mon = get_leader(); + list<Context*> retry; + for (map<uint64_t, RoutedRequest*>::iterator p = routed_requests.begin(); + p != routed_requests.end(); + ++p) { + RoutedRequest *rr = p->second; + + if (mon == rank) { + dout(10) << " requeue for self tid " << rr->tid << dendl; + rr->op->mark_event("retry routed request"); + retry.push_back(new C_RetryMessage(this, rr->op)); + if (rr->session) { + ceph_assert(rr->session->routed_request_tids.count(p->first)); + rr->session->routed_request_tids.erase(p->first); + } + delete rr; + } else { + auto q = rr->request_bl.cbegin(); + PaxosServiceMessage *req = + (PaxosServiceMessage *)decode_message(cct, 0, q); + rr->op->mark_event("resend forwarded message to leader"); + dout(10) << " resend to mon." << mon << " tid " << rr->tid << " " << *req + << dendl; + MForward *forward = new MForward(rr->tid, + req, + rr->con_features, + rr->session->caps); + req->put(); // forward takes its own ref; drop ours. + forward->client_type = rr->con->get_peer_type(); + forward->client_addrs = rr->con->get_peer_addrs(); + forward->client_socket_addr = rr->con->get_peer_socket_addr(); + forward->set_priority(req->get_priority()); + send_mon_message(forward, mon); + } + } + if (mon == rank) { + routed_requests.clear(); + finish_contexts(g_ceph_context, retry); + } +} + +void Monitor::remove_session(MonSession *s) +{ + dout(10) << "remove_session " << s << " " << s->name << " " << s->addrs + << " features 0x" << std::hex << s->con_features << std::dec << dendl; + ceph_assert(s->con); + ceph_assert(!s->closed); + for (set<uint64_t>::iterator p = s->routed_request_tids.begin(); + p != s->routed_request_tids.end(); + ++p) { + ceph_assert(routed_requests.count(*p)); + RoutedRequest *rr = routed_requests[*p]; + dout(10) << " dropping routed request " << rr->tid << dendl; + delete rr; + routed_requests.erase(*p); + } + s->routed_request_tids.clear(); + s->con->set_priv(nullptr); + session_map.remove_session(s); + logger->set(l_mon_num_sessions, session_map.get_size()); + logger->inc(l_mon_session_rm); +} + +void Monitor::remove_all_sessions() +{ + std::lock_guard l(session_map_lock); + while (!session_map.sessions.empty()) { + MonSession *s = session_map.sessions.front(); + remove_session(s); + logger->inc(l_mon_session_rm); + } + if (logger) + logger->set(l_mon_num_sessions, session_map.get_size()); +} + +void Monitor::send_mon_message(Message *m, int rank) +{ + messenger->send_to_mon(m, monmap->get_addrs(rank)); +} + +void Monitor::waitlist_or_zap_client(MonOpRequestRef op) +{ + /** + * Wait list the new session until we're in the quorum, assuming it's + * sufficiently new. + * tick() will periodically send them back through so we can send + * the client elsewhere if we don't think we're getting back in. + * + * But we whitelist a few sorts of messages: + * 1) Monitors can talk to us at any time, of course. + * 2) auth messages. It's unlikely to go through much faster, but + * it's possible we've just lost our quorum status and we want to take... + * 3) command messages. We want to accept these under all possible + * circumstances. + */ + Message *m = op->get_req(); + MonSession *s = op->get_session(); + ConnectionRef con = op->get_connection(); + utime_t too_old = ceph_clock_now(); + too_old -= g_ceph_context->_conf->mon_lease; + if (m->get_recv_stamp() > too_old && + con->is_connected()) { + dout(5) << "waitlisting message " << *m << dendl; + maybe_wait_for_quorum.push_back(new C_RetryMessage(this, op)); + op->mark_wait_for_quorum(); + } else { + dout(5) << "discarding message " << *m << " and sending client elsewhere" << dendl; + con->mark_down(); + // proxied sessions aren't registered and don't have a con; don't remove + // those. + if (!s->proxy_con) { + std::lock_guard l(session_map_lock); + remove_session(s); + } + op->mark_zap(); + } +} + +void Monitor::_ms_dispatch(Message *m) +{ + if (is_shutdown()) { + m->put(); + return; + } + + MonOpRequestRef op = op_tracker.create_request<MonOpRequest>(m); + bool src_is_mon = op->is_src_mon(); + op->mark_event("mon:_ms_dispatch"); + MonSession *s = op->get_session(); + if (s && s->closed) { + return; + } + + if (src_is_mon && s) { + ConnectionRef con = m->get_connection(); + if (con->get_messenger() && con->get_features() != s->con_features) { + // only update features if this is a non-anonymous connection + dout(10) << __func__ << " feature change for " << m->get_source_inst() + << " (was " << s->con_features + << ", now " << con->get_features() << ")" << dendl; + // connection features changed - recreate session. + if (s->con && s->con != con) { + dout(10) << __func__ << " connection for " << m->get_source_inst() + << " changed from session; mark down and replace" << dendl; + s->con->mark_down(); + } + if (s->item.is_on_list()) { + // forwarded messages' sessions are not in the sessions map and + // exist only while the op is being handled. + std::lock_guard l(session_map_lock); + remove_session(s); + } + s = nullptr; + } + } + + if (!s) { + // if the sender is not a monitor, make sure their first message for a + // session is an MAuth. If it is not, assume it's a stray message, + // and considering that we are creating a new session it is safe to + // assume that the sender hasn't authenticated yet, so we have no way + // of assessing whether we should handle it or not. + if (!src_is_mon && (m->get_type() != CEPH_MSG_AUTH && + m->get_type() != CEPH_MSG_MON_GET_MAP && + m->get_type() != CEPH_MSG_PING)) { + dout(1) << __func__ << " dropping stray message " << *m + << " from " << m->get_source_inst() << dendl; + return; + } + + ConnectionRef con = m->get_connection(); + { + std::lock_guard l(session_map_lock); + s = session_map.new_session(m->get_source(), + m->get_source_addrs(), + con.get()); + } + ceph_assert(s); + con->set_priv(RefCountedPtr{s, false}); + dout(10) << __func__ << " new session " << s << " " << *s + << " features 0x" << std::hex + << s->con_features << std::dec << dendl; + op->set_session(s); + + logger->set(l_mon_num_sessions, session_map.get_size()); + logger->inc(l_mon_session_add); + + if (src_is_mon) { + // give it monitor caps; the peer type has been authenticated + dout(5) << __func__ << " setting monitor caps on this connection" << dendl; + if (!s->caps.is_allow_all()) // but no need to repeatedly copy + s->caps = mon_caps; + s->authenticated = true; + } + } else { + dout(20) << __func__ << " existing session " << s << " for " << s->name + << dendl; + } + + ceph_assert(s); + + s->session_timeout = ceph_clock_now(); + s->session_timeout += g_conf()->mon_session_timeout; + + if (s->auth_handler) { + s->entity_name = s->auth_handler->get_entity_name(); + s->global_id = s->auth_handler->get_global_id(); + s->global_id_status = s->auth_handler->get_global_id_status(); + } + dout(20) << " entity_name " << s->entity_name + << " global_id " << s->global_id + << " (" << s->global_id_status + << ") caps " << s->caps.get_str() << dendl; + + if ((is_synchronizing() || + (!s->authenticated && !exited_quorum.is_zero())) && + !src_is_mon && + m->get_type() != CEPH_MSG_PING) { + waitlist_or_zap_client(op); + } else { + dispatch_op(op); + } + return; +} + +void Monitor::dispatch_op(MonOpRequestRef op) +{ + op->mark_event("mon:dispatch_op"); + MonSession *s = op->get_session(); + ceph_assert(s); + if (s->closed) { + dout(10) << " session closed, dropping " << op->get_req() << dendl; + return; + } + + /* we will consider the default type as being 'monitor' until proven wrong */ + op->set_type_monitor(); + /* deal with all messages that do not necessarily need caps */ + switch (op->get_req()->get_type()) { + // auth + case MSG_MON_GLOBAL_ID: + case CEPH_MSG_AUTH: + op->set_type_service(); + /* no need to check caps here */ + paxos_service[PAXOS_AUTH]->dispatch(op); + return; + + case CEPH_MSG_PING: + handle_ping(op); + return; + } + + if (!op->get_session()->authenticated) { + dout(5) << __func__ << " " << op->get_req()->get_source_inst() + << " is not authenticated, dropping " << *(op->get_req()) + << dendl; + return; + } + + // global_id_status == NONE: all sessions for auth_none and krb, + // mon <-> mon sessions (including proxied sessions) for cephx + ceph_assert(s->global_id_status == global_id_status_t::NONE || + s->global_id_status == global_id_status_t::NEW_OK || + s->global_id_status == global_id_status_t::NEW_NOT_EXPOSED || + s->global_id_status == global_id_status_t::RECLAIM_OK || + s->global_id_status == global_id_status_t::RECLAIM_INSECURE); + + // let mon_getmap through for "ping" (which doesn't reconnect) + // and "tell" (which reconnects but doesn't attempt to preserve + // its global_id and stays in NEW_NOT_EXPOSED, retrying until + // ->send_attempts reaches 0) + if (cct->_conf->auth_expose_insecure_global_id_reclaim && + s->global_id_status == global_id_status_t::NEW_NOT_EXPOSED && + op->get_req()->get_type() != CEPH_MSG_MON_GET_MAP) { + dout(5) << __func__ << " " << op->get_req()->get_source_inst() + << " may omit old_ticket on reconnects, discarding " + << *op->get_req() << " and forcing reconnect" << dendl; + ceph_assert(s->con && !s->proxy_con); + s->con->mark_down(); + { + std::lock_guard l(session_map_lock); + remove_session(s); + } + op->mark_zap(); + return; + } + + switch (op->get_req()->get_type()) { + case CEPH_MSG_MON_GET_MAP: + handle_mon_get_map(op); + return; + + case MSG_GET_CONFIG: + configmon()->handle_get_config(op); + return; + + case CEPH_MSG_MON_METADATA: + return handle_mon_metadata(op); + + case CEPH_MSG_MON_SUBSCRIBE: + /* FIXME: check what's being subscribed, filter accordingly */ + handle_subscribe(op); + return; + } + + /* well, maybe the op belongs to a service... */ + op->set_type_service(); + /* deal with all messages which caps should be checked somewhere else */ + switch (op->get_req()->get_type()) { + + // OSDs + case CEPH_MSG_MON_GET_OSDMAP: + case CEPH_MSG_POOLOP: + case MSG_OSD_BEACON: + case MSG_OSD_MARK_ME_DOWN: + case MSG_OSD_FULL: + case MSG_OSD_FAILURE: + case MSG_OSD_BOOT: + case MSG_OSD_ALIVE: + case MSG_OSD_PGTEMP: + case MSG_OSD_PG_CREATED: + case MSG_REMOVE_SNAPS: + case MSG_OSD_PG_READY_TO_MERGE: + paxos_service[PAXOS_OSDMAP]->dispatch(op); + return; + + // MDSs + case MSG_MDS_BEACON: + case MSG_MDS_OFFLOAD_TARGETS: + paxos_service[PAXOS_MDSMAP]->dispatch(op); + return; + + // Mgrs + case MSG_MGR_BEACON: + paxos_service[PAXOS_MGR]->dispatch(op); + return; + + // MgrStat + case MSG_MON_MGR_REPORT: + case CEPH_MSG_STATFS: + case MSG_GETPOOLSTATS: + paxos_service[PAXOS_MGRSTAT]->dispatch(op); + return; + + // log + case MSG_LOG: + paxos_service[PAXOS_LOG]->dispatch(op); + return; + + // handle_command() does its own caps checking + case MSG_MON_COMMAND: + op->set_type_command(); + handle_command(op); + return; + } + + /* nop, looks like it's not a service message; revert back to monitor */ + op->set_type_monitor(); + + /* messages we, the Monitor class, need to deal with + * but may be sent by clients. */ + + if (!op->get_session()->is_capable("mon", MON_CAP_R)) { + dout(5) << __func__ << " " << op->get_req()->get_source_inst() + << " not enough caps for " << *(op->get_req()) << " -- dropping" + << dendl; + return; + } + + switch (op->get_req()->get_type()) { + // misc + case CEPH_MSG_MON_GET_VERSION: + handle_get_version(op); + return; + } + + if (!op->is_src_mon()) { + dout(1) << __func__ << " unexpected monitor message from" + << " non-monitor entity " << op->get_req()->get_source_inst() + << " " << *(op->get_req()) << " -- dropping" << dendl; + return; + } + + /* messages that should only be sent by another monitor */ + switch (op->get_req()->get_type()) { + + case MSG_ROUTE: + handle_route(op); + return; + + case MSG_MON_PROBE: + handle_probe(op); + return; + + // Sync (i.e., the new slurp, but on steroids) + case MSG_MON_SYNC: + handle_sync(op); + return; + case MSG_MON_SCRUB: + handle_scrub(op); + return; + + /* log acks are sent from a monitor we sent the MLog to, and are + never sent by clients to us. */ + case MSG_LOGACK: + log_client.handle_log_ack((MLogAck*)op->get_req()); + return; + + // monmap + case MSG_MON_JOIN: + op->set_type_service(); + paxos_service[PAXOS_MONMAP]->dispatch(op); + return; + + // paxos + case MSG_MON_PAXOS: + { + op->set_type_paxos(); + MMonPaxos *pm = static_cast<MMonPaxos*>(op->get_req()); + if (!op->get_session()->is_capable("mon", MON_CAP_X)) { + //can't send these! + return; + } + + if (state == STATE_SYNCHRONIZING) { + // we are synchronizing. These messages would do us no + // good, thus just drop them and ignore them. + dout(10) << __func__ << " ignore paxos msg from " + << pm->get_source_inst() << dendl; + return; + } + + // sanitize + if (pm->epoch > get_epoch()) { + bootstrap(); + return; + } + if (pm->epoch != get_epoch()) { + return; + } + + paxos->dispatch(op); + } + return; + + // elector messages + case MSG_MON_ELECTION: + op->set_type_election(); + //check privileges here for simplicity + if (!op->get_session()->is_capable("mon", MON_CAP_X)) { + dout(0) << "MMonElection received from entity without enough caps!" + << op->get_session()->caps << dendl; + return;; + } + if (!is_probing() && !is_synchronizing()) { + elector.dispatch(op); + } + return; + + case MSG_FORWARD: + handle_forward(op); + return; + + case MSG_TIMECHECK: + dout(5) << __func__ << " ignoring " << op << dendl; + return; + case MSG_TIMECHECK2: + handle_timecheck(op); + return; + + case MSG_MON_HEALTH: + dout(5) << __func__ << " dropping deprecated message: " + << *op->get_req() << dendl; + break; + case MSG_MON_HEALTH_CHECKS: + op->set_type_service(); + paxos_service[PAXOS_HEALTH]->dispatch(op); + return; + } + dout(1) << "dropping unexpected " << *(op->get_req()) << dendl; + return; +} + +void Monitor::handle_ping(MonOpRequestRef op) +{ + MPing *m = static_cast<MPing*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + MPing *reply = new MPing; + bufferlist payload; + boost::scoped_ptr<Formatter> f(new JSONFormatter(true)); + f->open_object_section("pong"); + + get_health_status(false, f.get(), nullptr); + { + stringstream ss; + get_mon_status(f.get(), ss); + } + + f->close_section(); + stringstream ss; + f->flush(ss); + encode(ss.str(), payload); + reply->set_payload(payload); + dout(10) << __func__ << " reply payload len " << reply->get_payload().length() << dendl; + m->get_connection()->send_message(reply); +} + +void Monitor::timecheck_start() +{ + dout(10) << __func__ << dendl; + timecheck_cleanup(); + if (get_quorum_mon_features().contains_all( + ceph::features::mon::FEATURE_NAUTILUS)) { + timecheck_start_round(); + } +} + +void Monitor::timecheck_finish() +{ + dout(10) << __func__ << dendl; + timecheck_cleanup(); +} + +void Monitor::timecheck_start_round() +{ + dout(10) << __func__ << " curr " << timecheck_round << dendl; + ceph_assert(is_leader()); + + if (monmap->size() == 1) { + ceph_abort_msg("We are alone; this shouldn't have been scheduled!"); + return; + } + + if (timecheck_round % 2) { + dout(10) << __func__ << " there's a timecheck going on" << dendl; + utime_t curr_time = ceph_clock_now(); + double max = g_conf()->mon_timecheck_interval*3; + if (curr_time - timecheck_round_start < max) { + dout(10) << __func__ << " keep current round going" << dendl; + goto out; + } else { + dout(10) << __func__ + << " finish current timecheck and start new" << dendl; + timecheck_cancel_round(); + } + } + + ceph_assert(timecheck_round % 2 == 0); + timecheck_acks = 0; + timecheck_round ++; + timecheck_round_start = ceph_clock_now(); + dout(10) << __func__ << " new " << timecheck_round << dendl; + + timecheck(); +out: + dout(10) << __func__ << " setting up next event" << dendl; + timecheck_reset_event(); +} + +void Monitor::timecheck_finish_round(bool success) +{ + dout(10) << __func__ << " curr " << timecheck_round << dendl; + ceph_assert(timecheck_round % 2); + timecheck_round ++; + timecheck_round_start = utime_t(); + + if (success) { + ceph_assert(timecheck_waiting.empty()); + ceph_assert(timecheck_acks == quorum.size()); + timecheck_report(); + timecheck_check_skews(); + return; + } + + dout(10) << __func__ << " " << timecheck_waiting.size() + << " peers still waiting:"; + for (auto& p : timecheck_waiting) { + *_dout << " mon." << p.first; + } + *_dout << dendl; + timecheck_waiting.clear(); + + dout(10) << __func__ << " finished to " << timecheck_round << dendl; +} + +void Monitor::timecheck_cancel_round() +{ + timecheck_finish_round(false); +} + +void Monitor::timecheck_cleanup() +{ + timecheck_round = 0; + timecheck_acks = 0; + timecheck_round_start = utime_t(); + + if (timecheck_event) { + timer.cancel_event(timecheck_event); + timecheck_event = NULL; + } + timecheck_waiting.clear(); + timecheck_skews.clear(); + timecheck_latencies.clear(); + + timecheck_rounds_since_clean = 0; +} + +void Monitor::timecheck_reset_event() +{ + if (timecheck_event) { + timer.cancel_event(timecheck_event); + timecheck_event = NULL; + } + + double delay = + cct->_conf->mon_timecheck_skew_interval * timecheck_rounds_since_clean; + + if (delay <= 0 || delay > cct->_conf->mon_timecheck_interval) { + delay = cct->_conf->mon_timecheck_interval; + } + + dout(10) << __func__ << " delay " << delay + << " rounds_since_clean " << timecheck_rounds_since_clean + << dendl; + + timecheck_event = timer.add_event_after( + delay, + new C_MonContext(this, [this](int) { + timecheck_start_round(); + })); +} + +void Monitor::timecheck_check_skews() +{ + dout(10) << __func__ << dendl; + ceph_assert(is_leader()); + ceph_assert((timecheck_round % 2) == 0); + if (monmap->size() == 1) { + ceph_abort_msg("We are alone; we shouldn't have gotten here!"); + return; + } + ceph_assert(timecheck_latencies.size() == timecheck_skews.size()); + + bool found_skew = false; + for (auto& p : timecheck_skews) { + double abs_skew; + if (timecheck_has_skew(p.second, &abs_skew)) { + dout(10) << __func__ + << " " << p.first << " skew " << abs_skew << dendl; + found_skew = true; + } + } + + if (found_skew) { + ++timecheck_rounds_since_clean; + timecheck_reset_event(); + } else if (timecheck_rounds_since_clean > 0) { + dout(1) << __func__ + << " no clock skews found after " << timecheck_rounds_since_clean + << " rounds" << dendl; + // make sure the skews are really gone and not just a transient success + // this will run just once if not in the presence of skews again. + timecheck_rounds_since_clean = 1; + timecheck_reset_event(); + timecheck_rounds_since_clean = 0; + } + +} + +void Monitor::timecheck_report() +{ + dout(10) << __func__ << dendl; + ceph_assert(is_leader()); + ceph_assert((timecheck_round % 2) == 0); + if (monmap->size() == 1) { + ceph_abort_msg("We are alone; we shouldn't have gotten here!"); + return; + } + + ceph_assert(timecheck_latencies.size() == timecheck_skews.size()); + bool do_output = true; // only output report once + for (set<int>::iterator q = quorum.begin(); q != quorum.end(); ++q) { + if (monmap->get_name(*q) == name) + continue; + + MTimeCheck2 *m = new MTimeCheck2(MTimeCheck2::OP_REPORT); + m->epoch = get_epoch(); + m->round = timecheck_round; + + for (auto& it : timecheck_skews) { + double skew = it.second; + double latency = timecheck_latencies[it.first]; + + m->skews[it.first] = skew; + m->latencies[it.first] = latency; + + if (do_output) { + dout(25) << __func__ << " mon." << it.first + << " latency " << latency + << " skew " << skew << dendl; + } + } + do_output = false; + dout(10) << __func__ << " send report to mon." << *q << dendl; + send_mon_message(m, *q); + } +} + +void Monitor::timecheck() +{ + dout(10) << __func__ << dendl; + ceph_assert(is_leader()); + if (monmap->size() == 1) { + ceph_abort_msg("We are alone; we shouldn't have gotten here!"); + return; + } + ceph_assert(timecheck_round % 2 != 0); + + timecheck_acks = 1; // we ack ourselves + + dout(10) << __func__ << " start timecheck epoch " << get_epoch() + << " round " << timecheck_round << dendl; + + // we are at the eye of the storm; the point of reference + timecheck_skews[rank] = 0.0; + timecheck_latencies[rank] = 0.0; + + for (set<int>::iterator it = quorum.begin(); it != quorum.end(); ++it) { + if (monmap->get_name(*it) == name) + continue; + + utime_t curr_time = ceph_clock_now(); + timecheck_waiting[*it] = curr_time; + MTimeCheck2 *m = new MTimeCheck2(MTimeCheck2::OP_PING); + m->epoch = get_epoch(); + m->round = timecheck_round; + dout(10) << __func__ << " send " << *m << " to mon." << *it << dendl; + send_mon_message(m, *it); + } +} + +health_status_t Monitor::timecheck_status(ostringstream &ss, + const double skew_bound, + const double latency) +{ + health_status_t status = HEALTH_OK; + ceph_assert(latency >= 0); + + double abs_skew; + if (timecheck_has_skew(skew_bound, &abs_skew)) { + status = HEALTH_WARN; + ss << "clock skew " << abs_skew << "s" + << " > max " << g_conf()->mon_clock_drift_allowed << "s"; + } + + return status; +} + +void Monitor::handle_timecheck_leader(MonOpRequestRef op) +{ + MTimeCheck2 *m = static_cast<MTimeCheck2*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + /* handles PONG's */ + ceph_assert(m->op == MTimeCheck2::OP_PONG); + + int other = m->get_source().num(); + if (m->epoch < get_epoch()) { + dout(1) << __func__ << " got old timecheck epoch " << m->epoch + << " from " << other + << " curr " << get_epoch() + << " -- severely lagged? discard" << dendl; + return; + } + ceph_assert(m->epoch == get_epoch()); + + if (m->round < timecheck_round) { + dout(1) << __func__ << " got old round " << m->round + << " from " << other + << " curr " << timecheck_round << " -- discard" << dendl; + return; + } + + utime_t curr_time = ceph_clock_now(); + + ceph_assert(timecheck_waiting.count(other) > 0); + utime_t timecheck_sent = timecheck_waiting[other]; + timecheck_waiting.erase(other); + if (curr_time < timecheck_sent) { + // our clock was readjusted -- drop everything until it all makes sense. + dout(1) << __func__ << " our clock was readjusted --" + << " bump round and drop current check" + << dendl; + timecheck_cancel_round(); + return; + } + + /* update peer latencies */ + double latency = (double)(curr_time - timecheck_sent); + + if (timecheck_latencies.count(other) == 0) + timecheck_latencies[other] = latency; + else { + double avg_latency = ((timecheck_latencies[other]*0.8)+(latency*0.2)); + timecheck_latencies[other] = avg_latency; + } + + /* + * update skews + * + * some nasty thing goes on if we were to do 'a - b' between two utime_t, + * and 'a' happens to be lower than 'b'; so we use double instead. + * + * latency is always expected to be >= 0. + * + * delta, the difference between theirs timestamp and ours, may either be + * lower or higher than 0; will hardly ever be 0. + * + * The absolute skew is the absolute delta minus the latency, which is + * taken as a whole instead of an rtt given that there is some queueing + * and dispatch times involved and it's hard to assess how long exactly + * it took for the message to travel to the other side and be handled. So + * we call it a bounded skew, the worst case scenario. + * + * Now, to math! + * + * Given that the latency is always positive, we can establish that the + * bounded skew will be: + * + * 1. positive if the absolute delta is higher than the latency and + * delta is positive + * 2. negative if the absolute delta is higher than the latency and + * delta is negative. + * 3. zero if the absolute delta is lower than the latency. + * + * On 3. we make a judgement call and treat the skew as non-existent. + * This is because that, if the absolute delta is lower than the + * latency, then the apparently existing skew is nothing more than a + * side-effect of the high latency at work. + * + * This may not be entirely true though, as a severely skewed clock + * may be masked by an even higher latency, but with high latencies + * we probably have worse issues to deal with than just skewed clocks. + */ + ceph_assert(latency >= 0); + + double delta = ((double) m->timestamp) - ((double) curr_time); + double abs_delta = (delta > 0 ? delta : -delta); + double skew_bound = abs_delta - latency; + if (skew_bound < 0) + skew_bound = 0; + else if (delta < 0) + skew_bound = -skew_bound; + + ostringstream ss; + health_status_t status = timecheck_status(ss, skew_bound, latency); + if (status != HEALTH_OK) { + clog->health(status) << other << " " << ss.str(); + } + + dout(10) << __func__ << " from " << other << " ts " << m->timestamp + << " delta " << delta << " skew_bound " << skew_bound + << " latency " << latency << dendl; + + timecheck_skews[other] = skew_bound; + + timecheck_acks++; + if (timecheck_acks == quorum.size()) { + dout(10) << __func__ << " got pongs from everybody (" + << timecheck_acks << " total)" << dendl; + ceph_assert(timecheck_skews.size() == timecheck_acks); + ceph_assert(timecheck_waiting.empty()); + // everyone has acked, so bump the round to finish it. + timecheck_finish_round(); + } +} + +void Monitor::handle_timecheck_peon(MonOpRequestRef op) +{ + MTimeCheck2 *m = static_cast<MTimeCheck2*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + + ceph_assert(is_peon()); + ceph_assert(m->op == MTimeCheck2::OP_PING || m->op == MTimeCheck2::OP_REPORT); + + if (m->epoch != get_epoch()) { + dout(1) << __func__ << " got wrong epoch " + << "(ours " << get_epoch() + << " theirs: " << m->epoch << ") -- discarding" << dendl; + return; + } + + if (m->round < timecheck_round) { + dout(1) << __func__ << " got old round " << m->round + << " current " << timecheck_round + << " (epoch " << get_epoch() << ") -- discarding" << dendl; + return; + } + + timecheck_round = m->round; + + if (m->op == MTimeCheck2::OP_REPORT) { + ceph_assert((timecheck_round % 2) == 0); + timecheck_latencies.swap(m->latencies); + timecheck_skews.swap(m->skews); + return; + } + + ceph_assert((timecheck_round % 2) != 0); + MTimeCheck2 *reply = new MTimeCheck2(MTimeCheck2::OP_PONG); + utime_t curr_time = ceph_clock_now(); + reply->timestamp = curr_time; + reply->epoch = m->epoch; + reply->round = m->round; + dout(10) << __func__ << " send " << *m + << " to " << m->get_source_inst() << dendl; + m->get_connection()->send_message(reply); +} + +void Monitor::handle_timecheck(MonOpRequestRef op) +{ + MTimeCheck2 *m = static_cast<MTimeCheck2*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + + if (is_leader()) { + if (m->op != MTimeCheck2::OP_PONG) { + dout(1) << __func__ << " drop unexpected msg (not pong)" << dendl; + } else { + handle_timecheck_leader(op); + } + } else if (is_peon()) { + if (m->op != MTimeCheck2::OP_PING && m->op != MTimeCheck2::OP_REPORT) { + dout(1) << __func__ << " drop unexpected msg (not ping or report)" << dendl; + } else { + handle_timecheck_peon(op); + } + } else { + dout(1) << __func__ << " drop unexpected msg" << dendl; + } +} + +void Monitor::handle_subscribe(MonOpRequestRef op) +{ + MMonSubscribe *m = static_cast<MMonSubscribe*>(op->get_req()); + dout(10) << "handle_subscribe " << *m << dendl; + + bool reply = false; + + MonSession *s = op->get_session(); + ceph_assert(s); + + if (m->hostname.size()) { + s->remote_host = m->hostname; + } + + for (map<string,ceph_mon_subscribe_item>::iterator p = m->what.begin(); + p != m->what.end(); + ++p) { + if (p->first == "monmap" || p->first == "config") { + // these require no caps + } else if (!s->is_capable("mon", MON_CAP_R)) { + dout(5) << __func__ << " " << op->get_req()->get_source_inst() + << " not enough caps for " << *(op->get_req()) << " -- dropping" + << dendl; + continue; + } + + // if there are any non-onetime subscriptions, we need to reply to start the resubscribe timer + if ((p->second.flags & CEPH_SUBSCRIBE_ONETIME) == 0) + reply = true; + + // remove conflicting subscribes + if (logmon()->sub_name_to_id(p->first) >= 0) { + for (map<string, Subscription*>::iterator it = s->sub_map.begin(); + it != s->sub_map.end(); ) { + if (it->first != p->first && logmon()->sub_name_to_id(it->first) >= 0) { + std::lock_guard l(session_map_lock); + session_map.remove_sub((it++)->second); + } else { + ++it; + } + } + } + + { + std::lock_guard l(session_map_lock); + session_map.add_update_sub(s, p->first, p->second.start, + p->second.flags & CEPH_SUBSCRIBE_ONETIME, + m->get_connection()->has_feature(CEPH_FEATURE_INCSUBOSDMAP)); + } + + if (p->first.compare(0, 6, "mdsmap") == 0 || p->first.compare(0, 5, "fsmap") == 0) { + dout(10) << __func__ << ": MDS sub '" << p->first << "'" << dendl; + if ((int)s->is_capable("mds", MON_CAP_R)) { + Subscription *sub = s->sub_map[p->first]; + ceph_assert(sub != nullptr); + mdsmon()->check_sub(sub); + } + } else if (p->first == "osdmap") { + if ((int)s->is_capable("osd", MON_CAP_R)) { + if (s->osd_epoch > p->second.start) { + // client needs earlier osdmaps on purpose, so reset the sent epoch + s->osd_epoch = 0; + } + osdmon()->check_osdmap_sub(s->sub_map["osdmap"]); + } + } else if (p->first == "osd_pg_creates") { + if ((int)s->is_capable("osd", MON_CAP_W)) { + osdmon()->check_pg_creates_sub(s->sub_map["osd_pg_creates"]); + } + } else if (p->first == "monmap") { + monmon()->check_sub(s->sub_map[p->first]); + } else if (logmon()->sub_name_to_id(p->first) >= 0) { + logmon()->check_sub(s->sub_map[p->first]); + } else if (p->first == "mgrmap" || p->first == "mgrdigest") { + mgrmon()->check_sub(s->sub_map[p->first]); + } else if (p->first == "servicemap") { + mgrstatmon()->check_sub(s->sub_map[p->first]); + } else if (p->first == "config") { + configmon()->check_sub(s); + } + } + + if (reply) { + // we only need to reply if the client is old enough to think it + // has to send renewals. + ConnectionRef con = m->get_connection(); + if (!con->has_feature(CEPH_FEATURE_MON_STATEFUL_SUB)) + m->get_connection()->send_message(new MMonSubscribeAck( + monmap->get_fsid(), (int)g_conf()->mon_subscribe_interval)); + } + +} + +void Monitor::handle_get_version(MonOpRequestRef op) +{ + MMonGetVersion *m = static_cast<MMonGetVersion*>(op->get_req()); + dout(10) << "handle_get_version " << *m << dendl; + PaxosService *svc = NULL; + + MonSession *s = op->get_session(); + ceph_assert(s); + + if (!is_leader() && !is_peon()) { + dout(10) << " waiting for quorum" << dendl; + waitfor_quorum.push_back(new C_RetryMessage(this, op)); + goto out; + } + + if (m->what == "mdsmap") { + svc = mdsmon(); + } else if (m->what == "fsmap") { + svc = mdsmon(); + } else if (m->what == "osdmap") { + svc = osdmon(); + } else if (m->what == "monmap") { + svc = monmon(); + } else { + derr << "invalid map type " << m->what << dendl; + } + + if (svc) { + if (!svc->is_readable()) { + svc->wait_for_readable(op, new C_RetryMessage(this, op)); + goto out; + } + + MMonGetVersionReply *reply = new MMonGetVersionReply(); + reply->handle = m->handle; + reply->version = svc->get_last_committed(); + reply->oldest_version = svc->get_first_committed(); + reply->set_tid(m->get_tid()); + + m->get_connection()->send_message(reply); + } + out: + return; +} + +bool Monitor::ms_handle_reset(Connection *con) +{ + dout(10) << "ms_handle_reset " << con << " " << con->get_peer_addr() << dendl; + + // ignore lossless monitor sessions + if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) + return false; + + auto priv = con->get_priv(); + auto s = static_cast<MonSession*>(priv.get()); + if (!s) + return false; + + // break any con <-> session ref cycle + s->con->set_priv(nullptr); + + if (is_shutdown()) + return false; + + std::lock_guard l(lock); + + dout(10) << "reset/close on session " << s->name << " " << s->addrs << dendl; + if (!s->closed && s->item.is_on_list()) { + std::lock_guard l(session_map_lock); + remove_session(s); + } + return true; +} + +bool Monitor::ms_handle_refused(Connection *con) +{ + // just log for now... + dout(10) << "ms_handle_refused " << con << " " << con->get_peer_addr() << dendl; + return false; +} + +// ----- + +void Monitor::send_latest_monmap(Connection *con) +{ + bufferlist bl; + monmap->encode(bl, con->get_features()); + con->send_message(new MMonMap(bl)); +} + +void Monitor::handle_mon_get_map(MonOpRequestRef op) +{ + MMonGetMap *m = static_cast<MMonGetMap*>(op->get_req()); + dout(10) << "handle_mon_get_map" << dendl; + send_latest_monmap(m->get_connection().get()); +} + +void Monitor::handle_mon_metadata(MonOpRequestRef op) +{ + MMonMetadata *m = static_cast<MMonMetadata*>(op->get_req()); + if (is_leader()) { + dout(10) << __func__ << dendl; + update_mon_metadata(m->get_source().num(), std::move(m->data)); + } +} + +void Monitor::update_mon_metadata(int from, Metadata&& m) +{ + // NOTE: this is now for legacy (kraken or jewel) mons only. + pending_metadata[from] = std::move(m); + + MonitorDBStore::TransactionRef t = paxos->get_pending_transaction(); + bufferlist bl; + encode(pending_metadata, bl); + t->put(MONITOR_STORE_PREFIX, "last_metadata", bl); + paxos->trigger_propose(); +} + +int Monitor::load_metadata() +{ + bufferlist bl; + int r = store->get(MONITOR_STORE_PREFIX, "last_metadata", bl); + if (r) + return r; + auto it = bl.cbegin(); + decode(mon_metadata, it); + + pending_metadata = mon_metadata; + return 0; +} + +int Monitor::get_mon_metadata(int mon, Formatter *f, ostream& err) +{ + ceph_assert(f); + if (!mon_metadata.count(mon)) { + err << "mon." << mon << " not found"; + return -EINVAL; + } + const Metadata& m = mon_metadata[mon]; + for (Metadata::const_iterator p = m.begin(); p != m.end(); ++p) { + f->dump_string(p->first.c_str(), p->second); + } + return 0; +} + +void Monitor::count_metadata(const string& field, map<string,int> *out) +{ + for (auto& p : mon_metadata) { + auto q = p.second.find(field); + if (q == p.second.end()) { + (*out)["unknown"]++; + } else { + (*out)[q->second]++; + } + } +} + +void Monitor::count_metadata(const string& field, Formatter *f) +{ + map<string,int> by_val; + count_metadata(field, &by_val); + f->open_object_section(field.c_str()); + for (auto& p : by_val) { + f->dump_int(p.first.c_str(), p.second); + } + f->close_section(); +} + +int Monitor::print_nodes(Formatter *f, ostream& err) +{ + map<string, list<string> > mons; // hostname => mon + for (map<int, Metadata>::iterator it = mon_metadata.begin(); + it != mon_metadata.end(); ++it) { + const Metadata& m = it->second; + Metadata::const_iterator hostname = m.find("hostname"); + if (hostname == m.end()) { + // not likely though + continue; + } + mons[hostname->second].push_back(monmap->get_name(it->first)); + } + + dump_services(f, mons, "mon"); + return 0; +} + +// ---------------------------------------------- +// scrub + +int Monitor::scrub_start() +{ + dout(10) << __func__ << dendl; + ceph_assert(is_leader()); + + if (!scrub_result.empty()) { + clog->info() << "scrub already in progress"; + return -EBUSY; + } + + scrub_event_cancel(); + scrub_result.clear(); + scrub_state.reset(new ScrubState); + + scrub(); + return 0; +} + +int Monitor::scrub() +{ + ceph_assert(is_leader()); + ceph_assert(scrub_state); + + scrub_cancel_timeout(); + wait_for_paxos_write(); + scrub_version = paxos->get_version(); + + + // scrub all keys if we're the only monitor in the quorum + int32_t num_keys = + (quorum.size() == 1 ? -1 : cct->_conf->mon_scrub_max_keys); + + for (set<int>::iterator p = quorum.begin(); + p != quorum.end(); + ++p) { + if (*p == rank) + continue; + MMonScrub *r = new MMonScrub(MMonScrub::OP_SCRUB, scrub_version, + num_keys); + r->key = scrub_state->last_key; + send_mon_message(r, *p); + } + + // scrub my keys + bool r = _scrub(&scrub_result[rank], + &scrub_state->last_key, + &num_keys); + + scrub_state->finished = !r; + + // only after we got our scrub results do we really care whether the + // other monitors are late on their results. Also, this way we avoid + // triggering the timeout if we end up getting stuck in _scrub() for + // longer than the duration of the timeout. + scrub_reset_timeout(); + + if (quorum.size() == 1) { + ceph_assert(scrub_state->finished == true); + scrub_finish(); + } + return 0; +} + +void Monitor::handle_scrub(MonOpRequestRef op) +{ + MMonScrub *m = static_cast<MMonScrub*>(op->get_req()); + dout(10) << __func__ << " " << *m << dendl; + switch (m->op) { + case MMonScrub::OP_SCRUB: + { + if (!is_peon()) + break; + + wait_for_paxos_write(); + + if (m->version != paxos->get_version()) + break; + + MMonScrub *reply = new MMonScrub(MMonScrub::OP_RESULT, + m->version, + m->num_keys); + + reply->key = m->key; + _scrub(&reply->result, &reply->key, &reply->num_keys); + m->get_connection()->send_message(reply); + } + break; + + case MMonScrub::OP_RESULT: + { + if (!is_leader()) + break; + if (m->version != scrub_version) + break; + // reset the timeout each time we get a result + scrub_reset_timeout(); + + int from = m->get_source().num(); + ceph_assert(scrub_result.count(from) == 0); + scrub_result[from] = m->result; + + if (scrub_result.size() == quorum.size()) { + scrub_check_results(); + scrub_result.clear(); + if (scrub_state->finished) + scrub_finish(); + else + scrub(); + } + } + break; + } +} + +bool Monitor::_scrub(ScrubResult *r, + pair<string,string> *start, + int *num_keys) +{ + ceph_assert(r != NULL); + ceph_assert(start != NULL); + ceph_assert(num_keys != NULL); + + set<string> prefixes = get_sync_targets_names(); + prefixes.erase("paxos"); // exclude paxos, as this one may have extra states for proposals, etc. + + dout(10) << __func__ << " start (" << *start << ")" + << " num_keys " << *num_keys << dendl; + + MonitorDBStore::Synchronizer it = store->get_synchronizer(*start, prefixes); + + int scrubbed_keys = 0; + pair<string,string> last_key; + + while (it->has_next_chunk()) { + + if (*num_keys > 0 && scrubbed_keys == *num_keys) + break; + + pair<string,string> k = it->get_next_key(); + if (prefixes.count(k.first) == 0) + continue; + + if (cct->_conf->mon_scrub_inject_missing_keys > 0.0 && + (rand() % 10000 < cct->_conf->mon_scrub_inject_missing_keys*10000.0)) { + dout(10) << __func__ << " inject missing key, skipping (" << k << ")" + << dendl; + continue; + } + + bufferlist bl; + int err = store->get(k.first, k.second, bl); + ceph_assert(err == 0); + + uint32_t key_crc = bl.crc32c(0); + dout(30) << __func__ << " " << k << " bl " << bl.length() << " bytes" + << " crc " << key_crc << dendl; + r->prefix_keys[k.first]++; + if (r->prefix_crc.count(k.first) == 0) { + r->prefix_crc[k.first] = 0; + } + r->prefix_crc[k.first] = bl.crc32c(r->prefix_crc[k.first]); + + if (cct->_conf->mon_scrub_inject_crc_mismatch > 0.0 && + (rand() % 10000 < cct->_conf->mon_scrub_inject_crc_mismatch*10000.0)) { + dout(10) << __func__ << " inject failure at (" << k << ")" << dendl; + r->prefix_crc[k.first] += 1; + } + + ++scrubbed_keys; + last_key = k; + } + + dout(20) << __func__ << " last_key (" << last_key << ")" + << " scrubbed_keys " << scrubbed_keys + << " has_next " << it->has_next_chunk() << dendl; + + *start = last_key; + *num_keys = scrubbed_keys; + + return it->has_next_chunk(); +} + +void Monitor::scrub_check_results() +{ + dout(10) << __func__ << dendl; + + // compare + int errors = 0; + ScrubResult& mine = scrub_result[rank]; + for (map<int,ScrubResult>::iterator p = scrub_result.begin(); + p != scrub_result.end(); + ++p) { + if (p->first == rank) + continue; + if (p->second != mine) { + ++errors; + clog->error() << "scrub mismatch"; + clog->error() << " mon." << rank << " " << mine; + clog->error() << " mon." << p->first << " " << p->second; + } + } + if (!errors) + clog->debug() << "scrub ok on " << quorum << ": " << mine; +} + +inline void Monitor::scrub_timeout() +{ + dout(1) << __func__ << " restarting scrub" << dendl; + scrub_reset(); + scrub_start(); +} + +void Monitor::scrub_finish() +{ + dout(10) << __func__ << dendl; + scrub_reset(); + scrub_event_start(); +} + +void Monitor::scrub_reset() +{ + dout(10) << __func__ << dendl; + scrub_cancel_timeout(); + scrub_version = 0; + scrub_result.clear(); + scrub_state.reset(); +} + +inline void Monitor::scrub_update_interval(int secs) +{ + // we don't care about changes if we are not the leader. + // changes will be visible if we become the leader. + if (!is_leader()) + return; + + dout(1) << __func__ << " new interval = " << secs << dendl; + + // if scrub already in progress, all changes will already be visible during + // the next round. Nothing to do. + if (scrub_state != NULL) + return; + + scrub_event_cancel(); + scrub_event_start(); +} + +void Monitor::scrub_event_start() +{ + dout(10) << __func__ << dendl; + + if (scrub_event) + scrub_event_cancel(); + + if (cct->_conf->mon_scrub_interval <= 0) { + dout(1) << __func__ << " scrub event is disabled" + << " (mon_scrub_interval = " << cct->_conf->mon_scrub_interval + << ")" << dendl; + return; + } + + scrub_event = timer.add_event_after( + cct->_conf->mon_scrub_interval, + new C_MonContext(this, [this](int) { + scrub_start(); + })); +} + +void Monitor::scrub_event_cancel() +{ + dout(10) << __func__ << dendl; + if (scrub_event) { + timer.cancel_event(scrub_event); + scrub_event = NULL; + } +} + +inline void Monitor::scrub_cancel_timeout() +{ + if (scrub_timeout_event) { + timer.cancel_event(scrub_timeout_event); + scrub_timeout_event = NULL; + } +} + +void Monitor::scrub_reset_timeout() +{ + dout(15) << __func__ << " reset timeout event" << dendl; + scrub_cancel_timeout(); + scrub_timeout_event = timer.add_event_after( + g_conf()->mon_scrub_timeout, + new C_MonContext(this, [this](int) { + scrub_timeout(); + })); +} + +/************ TICK ***************/ +void Monitor::new_tick() +{ + timer.add_event_after(g_conf()->mon_tick_interval, new C_MonContext(this, [this](int) { + tick(); + })); +} + +void Monitor::tick() +{ + // ok go. + dout(11) << "tick" << dendl; + const utime_t now = ceph_clock_now(); + + // Check if we need to emit any delayed health check updated messages + if (is_leader()) { + const auto min_period = g_conf().get_val<int64_t>( + "mon_health_log_update_period"); + for (auto& svc : paxos_service) { + auto health = svc->get_health_checks(); + + for (const auto &i : health.checks) { + const std::string &code = i.first; + const std::string &summary = i.second.summary; + const health_status_t severity = i.second.severity; + + auto status_iter = health_check_log_times.find(code); + if (status_iter == health_check_log_times.end()) { + continue; + } + + auto &log_status = status_iter->second; + bool const changed = log_status.last_message != summary + || log_status.severity != severity; + + if (changed && now - log_status.updated_at > min_period) { + log_status.last_message = summary; + log_status.updated_at = now; + log_status.severity = severity; + + ostringstream ss; + ss << "Health check update: " << summary << " (" << code << ")"; + clog->health(severity) << ss.str(); + } + } + } + } + + + for (auto& svc : paxos_service) { + svc->tick(); + svc->maybe_trim(); + } + + // trim sessions + { + std::lock_guard l(session_map_lock); + auto p = session_map.sessions.begin(); + + bool out_for_too_long = (!exited_quorum.is_zero() && + now > (exited_quorum + 2*g_conf()->mon_lease)); + + while (!p.end()) { + MonSession *s = *p; + ++p; + + // don't trim monitors + if (s->name.is_mon()) + continue; + + if (s->session_timeout < now && s->con) { + // check keepalive, too + s->session_timeout = s->con->get_last_keepalive(); + s->session_timeout += g_conf()->mon_session_timeout; + } + if (s->session_timeout < now) { + dout(10) << " trimming session " << s->con << " " << s->name + << " " << s->addrs + << " (timeout " << s->session_timeout + << " < now " << now << ")" << dendl; + } else if (out_for_too_long) { + // boot the client Session because we've taken too long getting back in + dout(10) << " trimming session " << s->con << " " << s->name + << " because we've been out of quorum too long" << dendl; + } else { + continue; + } + + s->con->mark_down(); + remove_session(s); + logger->inc(l_mon_session_trim); + } + } + sync_trim_providers(); + + if (!maybe_wait_for_quorum.empty()) { + finish_contexts(g_ceph_context, maybe_wait_for_quorum); + } + + if (is_leader() && paxos->is_active() && fingerprint.is_zero()) { + // this is only necessary on upgraded clusters. + MonitorDBStore::TransactionRef t = paxos->get_pending_transaction(); + prepare_new_fingerprint(t); + paxos->trigger_propose(); + } + + mgr_client.update_daemon_health(get_health_metrics()); + new_tick(); +} + +vector<DaemonHealthMetric> Monitor::get_health_metrics() +{ + vector<DaemonHealthMetric> metrics; + + utime_t oldest_secs; + const utime_t now = ceph_clock_now(); + auto too_old = now; + too_old -= g_conf().get_val<std::chrono::seconds>("mon_op_complaint_time").count(); + int slow = 0; + TrackedOpRef oldest_op; + auto count_slow_ops = [&](TrackedOp& op) { + if (op.get_initiated() < too_old) { + slow++; + if (!oldest_op || op.get_initiated() < oldest_op->get_initiated()) { + oldest_op = &op; + } + return true; + } else { + return false; + } + }; + if (op_tracker.visit_ops_in_flight(&oldest_secs, count_slow_ops)) { + if (slow) { + derr << __func__ << " reporting " << slow << " slow ops, oldest is " + << oldest_op->get_desc() << dendl; + } + metrics.emplace_back(daemon_metric::SLOW_OPS, slow, oldest_secs); + } else { + metrics.emplace_back(daemon_metric::SLOW_OPS, 0, 0); + } + return metrics; +} + +void Monitor::prepare_new_fingerprint(MonitorDBStore::TransactionRef t) +{ + uuid_d nf; + nf.generate_random(); + dout(10) << __func__ << " proposing cluster_fingerprint " << nf << dendl; + + bufferlist bl; + encode(nf, bl); + t->put(MONITOR_NAME, "cluster_fingerprint", bl); +} + +int Monitor::check_fsid() +{ + bufferlist ebl; + int r = store->get(MONITOR_NAME, "cluster_uuid", ebl); + if (r == -ENOENT) + return r; + ceph_assert(r == 0); + + string es(ebl.c_str(), ebl.length()); + + // only keep the first line + size_t pos = es.find_first_of('\n'); + if (pos != string::npos) + es.resize(pos); + + dout(10) << "check_fsid cluster_uuid contains '" << es << "'" << dendl; + uuid_d ondisk; + if (!ondisk.parse(es.c_str())) { + derr << "error: unable to parse uuid" << dendl; + return -EINVAL; + } + + if (monmap->get_fsid() != ondisk) { + derr << "error: cluster_uuid file exists with value " << ondisk + << ", != our uuid " << monmap->get_fsid() << dendl; + return -EEXIST; + } + + return 0; +} + +int Monitor::write_fsid() +{ + auto t(std::make_shared<MonitorDBStore::Transaction>()); + write_fsid(t); + int r = store->apply_transaction(t); + return r; +} + +int Monitor::write_fsid(MonitorDBStore::TransactionRef t) +{ + ostringstream ss; + ss << monmap->get_fsid() << "\n"; + string us = ss.str(); + + bufferlist b; + b.append(us); + + t->put(MONITOR_NAME, "cluster_uuid", b); + return 0; +} + +/* + * this is the closest thing to a traditional 'mkfs' for ceph. + * initialize the monitor state machines to their initial values. + */ +int Monitor::mkfs(bufferlist& osdmapbl) +{ + auto t(std::make_shared<MonitorDBStore::Transaction>()); + + // verify cluster fsid + int r = check_fsid(); + if (r < 0 && r != -ENOENT) + return r; + + bufferlist magicbl; + magicbl.append(CEPH_MON_ONDISK_MAGIC); + magicbl.append("\n"); + t->put(MONITOR_NAME, "magic", magicbl); + + + features = get_initial_supported_features(); + write_features(t); + + // save monmap, osdmap, keyring. + bufferlist monmapbl; + monmap->encode(monmapbl, CEPH_FEATURES_ALL); + monmap->set_epoch(0); // must be 0 to avoid confusing first MonmapMonitor::update_from_paxos() + t->put("mkfs", "monmap", monmapbl); + + if (osdmapbl.length()) { + // make sure it's a valid osdmap + try { + OSDMap om; + om.decode(osdmapbl); + } + catch (buffer::error& e) { + derr << "error decoding provided osdmap: " << e.what() << dendl; + return -EINVAL; + } + t->put("mkfs", "osdmap", osdmapbl); + } + + if (is_keyring_required()) { + KeyRing keyring; + string keyring_filename; + + r = ceph_resolve_file_search(g_conf()->keyring, keyring_filename); + if (r) { + derr << "unable to find a keyring file on " << g_conf()->keyring + << ": " << cpp_strerror(r) << dendl; + if (g_conf()->key != "") { + string keyring_plaintext = "[mon.]\n\tkey = " + g_conf()->key + + "\n\tcaps mon = \"allow *\"\n"; + bufferlist bl; + bl.append(keyring_plaintext); + try { + auto i = bl.cbegin(); + keyring.decode_plaintext(i); + } + catch (const buffer::error& e) { + derr << "error decoding keyring " << keyring_plaintext + << ": " << e.what() << dendl; + return -EINVAL; + } + } else { + return -ENOENT; + } + } else { + r = keyring.load(g_ceph_context, keyring_filename); + if (r < 0) { + derr << "unable to load initial keyring " << g_conf()->keyring << dendl; + return r; + } + } + + // put mon. key in external keyring; seed with everything else. + extract_save_mon_key(keyring); + + bufferlist keyringbl; + keyring.encode_plaintext(keyringbl); + t->put("mkfs", "keyring", keyringbl); + } + write_fsid(t); + store->apply_transaction(t); + + return 0; +} + +int Monitor::write_default_keyring(bufferlist& bl) +{ + ostringstream os; + os << g_conf()->mon_data << "/keyring"; + + int err = 0; + int fd = ::open(os.str().c_str(), O_WRONLY|O_CREAT|O_CLOEXEC, 0600); + if (fd < 0) { + err = -errno; + dout(0) << __func__ << " failed to open " << os.str() + << ": " << cpp_strerror(err) << dendl; + return err; + } + + err = bl.write_fd(fd); + if (!err) + ::fsync(fd); + VOID_TEMP_FAILURE_RETRY(::close(fd)); + + return err; +} + +void Monitor::extract_save_mon_key(KeyRing& keyring) +{ + EntityName mon_name; + mon_name.set_type(CEPH_ENTITY_TYPE_MON); + EntityAuth mon_key; + if (keyring.get_auth(mon_name, mon_key)) { + dout(10) << "extract_save_mon_key moving mon. key to separate keyring" << dendl; + KeyRing pkey; + pkey.add(mon_name, mon_key); + bufferlist bl; + pkey.encode_plaintext(bl); + write_default_keyring(bl); + keyring.remove(mon_name); + } +} + +// AuthClient methods -- for mon <-> mon communication +int Monitor::get_auth_request( + Connection *con, + AuthConnectionMeta *auth_meta, + uint32_t *method, + vector<uint32_t> *preferred_modes, + bufferlist *out) +{ + std::scoped_lock l(auth_lock); + if (con->get_peer_type() != CEPH_ENTITY_TYPE_MON && + con->get_peer_type() != CEPH_ENTITY_TYPE_MGR) { + return -EACCES; + } + AuthAuthorizer *auth; + if (!ms_get_authorizer(con->get_peer_type(), &auth)) { + return -EACCES; + } + auth_meta->authorizer.reset(auth); + auth_registry.get_supported_modes(con->get_peer_type(), + auth->protocol, + preferred_modes); + *method = auth->protocol; + *out = auth->bl; + return 0; +} + +int Monitor::handle_auth_reply_more( + Connection *con, + AuthConnectionMeta *auth_meta, + const bufferlist& bl, + bufferlist *reply) +{ + std::scoped_lock l(auth_lock); + if (!auth_meta->authorizer) { + derr << __func__ << " no authorizer?" << dendl; + return -EACCES; + } + auth_meta->authorizer->add_challenge(cct, bl); + *reply = auth_meta->authorizer->bl; + return 0; +} + +int Monitor::handle_auth_done( + Connection *con, + AuthConnectionMeta *auth_meta, + uint64_t global_id, + uint32_t con_mode, + const bufferlist& bl, + CryptoKey *session_key, + std::string *connection_secret) +{ + std::scoped_lock l(auth_lock); + // verify authorizer reply + auto p = bl.begin(); + if (!auth_meta->authorizer->verify_reply(p, connection_secret)) { + dout(0) << __func__ << " failed verifying authorizer reply" << dendl; + return -EACCES; + } + auth_meta->session_key = auth_meta->authorizer->session_key; + return 0; +} + +int Monitor::handle_auth_bad_method( + Connection *con, + AuthConnectionMeta *auth_meta, + uint32_t old_auth_method, + int result, + const std::vector<uint32_t>& allowed_methods, + const std::vector<uint32_t>& allowed_modes) +{ + derr << __func__ << " hmm, they didn't like " << old_auth_method + << " result " << cpp_strerror(result) << dendl; + return -EACCES; +} + +bool Monitor::ms_get_authorizer(int service_id, AuthAuthorizer **authorizer) +{ + dout(10) << "ms_get_authorizer for " << ceph_entity_type_name(service_id) + << dendl; + + if (is_shutdown()) + return false; + + // we only connect to other monitors and mgr; every else connects to us. + if (service_id != CEPH_ENTITY_TYPE_MON && + service_id != CEPH_ENTITY_TYPE_MGR) + return false; + + if (!auth_cluster_required.is_supported_auth(CEPH_AUTH_CEPHX)) { + // auth_none + dout(20) << __func__ << " building auth_none authorizer" << dendl; + AuthNoneClientHandler handler{g_ceph_context}; + handler.set_global_id(0); + *authorizer = handler.build_authorizer(service_id); + return true; + } + + CephXServiceTicketInfo auth_ticket_info; + CephXSessionAuthInfo info; + int ret; + + EntityName name; + name.set_type(CEPH_ENTITY_TYPE_MON); + auth_ticket_info.ticket.name = name; + auth_ticket_info.ticket.global_id = 0; + + if (service_id == CEPH_ENTITY_TYPE_MON) { + // mon to mon authentication uses the private monitor shared key and not the + // rotating key + CryptoKey secret; + if (!keyring.get_secret(name, secret) && + !key_server.get_secret(name, secret)) { + dout(0) << " couldn't get secret for mon service from keyring or keyserver" + << dendl; + stringstream ss, ds; + int err = key_server.list_secrets(ds); + if (err < 0) + ss << "no installed auth entries!"; + else + ss << "installed auth entries:"; + dout(0) << ss.str() << "\n" << ds.str() << dendl; + return false; + } + + ret = key_server.build_session_auth_info( + service_id, auth_ticket_info.ticket, secret, (uint64_t)-1, info); + if (ret < 0) { + dout(0) << __func__ << " failed to build mon session_auth_info " + << cpp_strerror(ret) << dendl; + return false; + } + } else if (service_id == CEPH_ENTITY_TYPE_MGR) { + // mgr + ret = key_server.build_session_auth_info( + service_id, auth_ticket_info.ticket, info); + if (ret < 0) { + derr << __func__ << " failed to build mgr service session_auth_info " + << cpp_strerror(ret) << dendl; + return false; + } + } else { + ceph_abort(); // see check at top of fn + } + + CephXTicketBlob blob; + if (!cephx_build_service_ticket_blob(cct, info, blob)) { + dout(0) << "ms_get_authorizer failed to build service ticket" << dendl; + return false; + } + bufferlist ticket_data; + encode(blob, ticket_data); + + auto iter = ticket_data.cbegin(); + CephXTicketHandler handler(g_ceph_context, service_id); + decode(handler.ticket, iter); + + handler.session_key = info.session_key; + + *authorizer = handler.build_authorizer(0); + + return true; +} + +KeyStore *Monitor::ms_get_auth1_authorizer_keystore() +{ + return &keyring; +} + +int Monitor::handle_auth_request( + Connection *con, + AuthConnectionMeta *auth_meta, + bool more, + uint32_t auth_method, + const bufferlist &payload, + bufferlist *reply) +{ + std::scoped_lock l(auth_lock); + + // NOTE: be careful, the Connection hasn't fully negotiated yet, so + // e.g., peer_features, peer_addrs, and others are still unknown. + + dout(10) << __func__ << " con " << con << (more ? " (more)":" (start)") + << " method " << auth_method + << " payload " << payload.length() + << dendl; + if (!more) { + auth_meta->auth_mode = payload[0]; + } + + if (auth_meta->auth_mode >= AUTH_MODE_AUTHORIZER && + auth_meta->auth_mode <= AUTH_MODE_AUTHORIZER_MAX) { + AuthAuthorizeHandler *ah = get_auth_authorize_handler(con->get_peer_type(), + auth_method); + if (!ah) { + lderr(cct) << __func__ << " no AuthAuthorizeHandler found for auth method " + << auth_method << dendl; + return -EOPNOTSUPP; + } + bool was_challenge = (bool)auth_meta->authorizer_challenge; + bool isvalid = ah->verify_authorizer( + cct, + &keyring, + payload, + auth_meta->get_connection_secret_length(), + reply, + &con->peer_name, + &con->peer_global_id, + &con->peer_caps_info, + &auth_meta->session_key, + &auth_meta->connection_secret, + &auth_meta->authorizer_challenge); + if (isvalid) { + ms_handle_authentication(con); + return 1; + } + if (!more && !was_challenge && auth_meta->authorizer_challenge) { + return 0; + } + dout(10) << __func__ << " bad authorizer on " << con << dendl; + return -EACCES; + } else if (auth_meta->auth_mode < AUTH_MODE_MON || + auth_meta->auth_mode > AUTH_MODE_MON_MAX) { + derr << __func__ << " unrecognized auth mode " << auth_meta->auth_mode + << dendl; + return -EACCES; + } + + // wait until we've formed an initial quorum on mkfs so that we have + // the initial keys (e.g., client.admin). + if (authmon()->get_last_committed() == 0) { + dout(10) << __func__ << " haven't formed initial quorum, EBUSY" << dendl; + return -EBUSY; + } + + RefCountedPtr priv; + MonSession *s; + int32_t r = 0; + auto p = payload.begin(); + if (!more) { + if (con->get_priv()) { + return -EACCES; // wtf + } + + // handler? + unique_ptr<AuthServiceHandler> auth_handler{get_auth_service_handler( + auth_method, g_ceph_context, &key_server)}; + if (!auth_handler) { + dout(1) << __func__ << " auth_method " << auth_method << " not supported" + << dendl; + return -EOPNOTSUPP; + } + + uint8_t mode; + EntityName entity_name; + + try { + decode(mode, p); + if (mode < AUTH_MODE_MON || + mode > AUTH_MODE_MON_MAX) { + dout(1) << __func__ << " invalid mode " << (int)mode << dendl; + return -EACCES; + } + assert(mode >= AUTH_MODE_MON && mode <= AUTH_MODE_MON_MAX); + decode(entity_name, p); + decode(con->peer_global_id, p); + } catch (buffer::error& e) { + dout(1) << __func__ << " failed to decode, " << e.what() << dendl; + return -EACCES; + } + + // supported method? + if (entity_name.get_type() == CEPH_ENTITY_TYPE_MON || + entity_name.get_type() == CEPH_ENTITY_TYPE_OSD || + entity_name.get_type() == CEPH_ENTITY_TYPE_MDS || + entity_name.get_type() == CEPH_ENTITY_TYPE_MGR) { + if (!auth_cluster_required.is_supported_auth(auth_method)) { + dout(10) << __func__ << " entity " << entity_name << " method " + << auth_method << " not among supported " + << auth_cluster_required.get_supported_set() << dendl; + return -EOPNOTSUPP; + } + } else { + if (!auth_service_required.is_supported_auth(auth_method)) { + dout(10) << __func__ << " entity " << entity_name << " method " + << auth_method << " not among supported " + << auth_cluster_required.get_supported_set() << dendl; + return -EOPNOTSUPP; + } + } + + // for msgr1 we would do some weirdness here to ensure signatures + // are supported by the client if we require it. for msgr2 that + // is not necessary. + + bool is_new_global_id = false; + if (!con->peer_global_id) { + con->peer_global_id = authmon()->_assign_global_id(); + if (!con->peer_global_id) { + dout(1) << __func__ << " failed to assign global_id" << dendl; + return -EBUSY; + } + is_new_global_id = true; + } + + // set up partial session + s = new MonSession(con); + s->auth_handler = auth_handler.release(); + con->set_priv(RefCountedPtr{s, false}); + + r = s->auth_handler->start_session( + entity_name, + con->peer_global_id, + is_new_global_id, + reply, + &con->peer_caps_info); + } else { + priv = con->get_priv(); + if (!priv) { + // this can happen if the async ms_handle_reset event races with + // the unlocked call into handle_auth_request + return -EACCES; + } + s = static_cast<MonSession*>(priv.get()); + r = s->auth_handler->handle_request( + p, + auth_meta->get_connection_secret_length(), + reply, + &con->peer_caps_info, + &auth_meta->session_key, + &auth_meta->connection_secret); + } + if (r > 0 && + !s->authenticated) { + ms_handle_authentication(con); + } + + dout(30) << " r " << r << " reply:\n"; + reply->hexdump(*_dout); + *_dout << dendl; + return r; +} + +void Monitor::ms_handle_accept(Connection *con) +{ + auto priv = con->get_priv(); + MonSession *s = static_cast<MonSession*>(priv.get()); + if (!s) { + // legacy protocol v1? + dout(10) << __func__ << " con " << con << " no session" << dendl; + return; + } + + if (s->item.is_on_list()) { + dout(10) << __func__ << " con " << con << " session " << s + << " already on list" << dendl; + } else { + dout(10) << __func__ << " con " << con << " session " << s + << " registering session for " + << con->get_peer_addrs() << dendl; + s->_ident(entity_name_t(con->get_peer_type(), con->get_peer_id()), + con->get_peer_addrs()); + std::lock_guard l(session_map_lock); + session_map.add_session(s); + } +} + +int Monitor::ms_handle_authentication(Connection *con) +{ + if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) { + // mon <-> mon connections need no Session, and setting one up + // creates an awkward ref cycle between Session and Connection. + return 1; + } + + auto priv = con->get_priv(); + MonSession *s = static_cast<MonSession*>(priv.get()); + if (!s) { + // must be msgr2, otherwise dispatch would have set up the session. + s = session_map.new_session( + entity_name_t(con->get_peer_type(), -1), // we don't know yet + con->get_peer_addrs(), + con); + assert(s); + dout(10) << __func__ << " adding session " << s << " to con " << con + << dendl; + con->set_priv(s); + logger->set(l_mon_num_sessions, session_map.get_size()); + logger->inc(l_mon_session_add); + } + dout(10) << __func__ << " session " << s << " con " << con + << " addr " << s->con->get_peer_addr() + << " " << *s << dendl; + + AuthCapsInfo &caps_info = con->get_peer_caps_info(); + int ret = 0; + if (caps_info.allow_all) { + s->caps.set_allow_all(); + s->authenticated = true; + ret = 1; + } + if (caps_info.caps.length()) { + bufferlist::const_iterator p = caps_info.caps.cbegin(); + string str; + try { + decode(str, p); + } catch (const buffer::error &err) { + derr << __func__ << " corrupt cap data for " << con->get_peer_entity_name() + << " in auth db" << dendl; + str.clear(); + ret = -EPERM; + } + if (ret >= 0) { + if (s->caps.parse(str, NULL)) { + s->authenticated = true; + ret = 1; + } else { + derr << __func__ << " unparseable caps '" << str << "' for " + << con->get_peer_entity_name() << dendl; + ret = -EPERM; + } + } + } + + return ret; +} |