+++ /dev/null
-// -*- 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 <sstream>
-#include <stdlib.h>
-#include <signal.h>
-#include <limits.h>
-#include <cstring>
-#include <boost/scope_exit.hpp>
-#include <boost/algorithm/string/predicate.hpp>
-
-#include "Monitor.h"
-#include "common/version.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/MMonHealth.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/MStatfs.h"
-
-#include "messages/MMonSubscribe.h"
-#include "messages/MMonSubscribeAck.h"
-
-#include "messages/MAuthReply.h"
-
-#include "messages/MTimeCheck.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 "PGMonitor.h"
-#include "LogMonitor.h"
-#include "AuthMonitor.h"
-#include "MgrMonitor.h"
-#include "MgrStatMonitor.h"
-#include "mon/QuorumService.h"
-#include "mon/OldHealthMonitor.h"
-#include "mon/HealthMonitor.h"
-#include "mon/ConfigKeyService.h"
-#include "common/config.h"
-#include "common/cmdparse.h"
-#include "include/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, avail) \
- {parsesig, helptext, modulename, req_perms, avail, FLAG(NONE)},
-#define COMMAND_WITH_FLAG(parsesig, helptext, modulename, req_perms, avail, flags) \
- {parsesig, helptext, modulename, req_perms, avail, flags},
-MonCommand mon_commands[] = {
-#include <mon/MonCommands.h>
-};
-MonCommand pgmonitor_commands[] = {
-#include <mon/PGMonitorCommands.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_),
- 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),
- pgservice(nullptr),
- store(s),
-
- state(STATE_PROBING),
-
- 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, true, 1)
-{
- clog = log_client.create_channel(CLOG_CHANNEL_CLUSTER);
- audit_clog = log_client.create_channel(CLOG_CHANNEL_AUDIT);
-
- update_log_clients();
-
- paxos = new Paxos(this, "paxos");
-
- paxos_service[PAXOS_MDSMAP] = new MDSMonitor(this, paxos, "mdsmap");
- paxos_service[PAXOS_MONMAP] = new MonmapMonitor(this, paxos, "monmap");
- paxos_service[PAXOS_OSDMAP] = new OSDMonitor(cct, this, paxos, "osdmap");
- paxos_service[PAXOS_PGMAP] = new PGMonitor(this, paxos, "pgmap");
- paxos_service[PAXOS_LOG] = new LogMonitor(this, paxos, "logm");
- paxos_service[PAXOS_AUTH] = new AuthMonitor(this, paxos, "auth");
- paxos_service[PAXOS_MGR] = new MgrMonitor(this, paxos, "mgr");
- paxos_service[PAXOS_MGRSTAT] = new MgrStatMonitor(this, paxos, "mgrstat");
- paxos_service[PAXOS_HEALTH] = new HealthMonitor(this, paxos, "health");
-
- health_monitor = new OldHealthMonitor(this);
- config_key_service = new ConfigKeyService(this, paxos);
-
- mon_caps = new MonCap();
- bool r = mon_caps->parse("allow *", NULL);
- assert(r);
-
- exited_quorum = ceph_clock_now();
-
- // prepare local commands
- local_mon_commands.resize(ARRAY_SIZE(mon_commands));
- for (unsigned i = 0; i < ARRAY_SIZE(mon_commands); ++i) {
- local_mon_commands[i] = mon_commands[i];
- }
- MonCommand::encode_vector(local_mon_commands, local_mon_commands_bl);
-
- local_upgrading_mon_commands = local_mon_commands;
- for (unsigned i = 0; i < ARRAY_SIZE(pgmonitor_commands); ++i) {
- local_upgrading_mon_commands.push_back(pgmonitor_commands[i]);
- }
- MonCommand::encode_vector(local_upgrading_mon_commands,
- local_upgrading_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;
-
- // note: OSDMonitor may update this based on the luminous flag.
- pgservice = mgrstatmon()->get_pg_stat_service();
-}
-
-Monitor::~Monitor()
-{
- for (vector<PaxosService*>::iterator p = paxos_service.begin(); p != paxos_service.end(); ++p)
- delete *p;
- delete health_monitor;
- delete config_key_service;
- delete paxos;
- assert(session_map.sessions.empty());
- delete mon_caps;
-}
-
-
-class AdminHook : public AdminSocketHook {
- Monitor *mon;
-public:
- explicit AdminHook(Monitor *m) : mon(m) {}
- bool call(std::string command, cmdmap_t& cmdmap, std::string format,
- bufferlist& out) override {
- stringstream ss;
- mon->do_admin_command(command, cmdmap, format, ss);
- out.append(ss);
- return true;
- }
-};
-
-void Monitor::do_admin_command(string command, cmdmap_t& cmdmap, string format,
- ostream& ss)
-{
- Mutex::Locker l(lock);
-
- boost::scoped_ptr<Formatter> f(Formatter::create(format));
-
- string args;
- for (cmdmap_t::iterator 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) {
- 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_stream("session") << *p;
- }
- f->close_section();
- f->flush(ss);
- }
-
- } else {
- assert(0 == "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)
-{
- 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);
- 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 {
- bufferlist::iterator it = featuresbl.begin();
- 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",
- NULL
- };
- return KEYS;
-}
-
-void Monitor::handle_conf_change(const struct md_config_t *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")) {
- health_to_clog_update_conf(changed);
- }
-
- if (changed.count("mon_scrub_interval")) {
- scrub_update_interval(conf->mon_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;
- }
-
- 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);
- }
-
- 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");
- pcb.add_u64(l_cluster_osd_bytes_used, "osd_bytes_used", "Used space");
- pcb.add_u64(l_cluster_osd_bytes_avail, "osd_bytes_avail", "Available space");
- 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");
- pcb.add_u64(l_cluster_num_mds_up, "num_mds_up", "MDSs that are up");
- pcb.add_u64(l_cluster_num_mds_in, "num_mds_in", "MDS in state \"in\" (they are in cluster)");
- pcb.add_u64(l_cluster_num_mds_failed, "num_mds_failed", "Failed MDS");
- pcb.add_u64(l_cluster_mds_epoch, "mds_epoch", "Current epoch of MDS map");
- 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_myaddr(),
- &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();
- health_monitor->init();
-
- 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;
- bufferlist::iterator p = bl.begin();
- ::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");
- assert(r == 0);
- r = admin_socket->register_command("quorum_status", "quorum_status",
- admin_hook, "show current quorum status");
- 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");
- 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");
- assert(r == 0);
- r = admin_socket->register_command("quorum enter", "quorum enter",
- admin_hook,
- "force monitor back into quorum");
- assert(r == 0);
- r = admin_socket->register_command("quorum exit", "quorum exit",
- admin_hook,
- "force monitor out of the quorum");
- assert(r == 0);
- r = admin_socket->register_command("ops",
- "ops",
- admin_hook,
- "show the ops currently in flight");
- assert(r == 0);
- r = admin_socket->register_command("sessions",
- "sessions",
- admin_hook,
- "list existing sessions");
- assert(r == 0);
-
- lock.Lock();
-
- // add ourselves as a conf observer
- g_conf->add_observer(this);
-
- lock.Unlock();
- return 0;
-}
-
-int Monitor::init()
-{
- dout(2) << "init" << dendl;
- Mutex::Locker l(lock);
-
- finisher.start();
-
- // start ticker
- timer.init();
- new_tick();
-
- cpu_tp.start();
-
- // i'm ready!
- messenger->add_dispatcher_tail(this);
-
- mgr_client.init();
- mgr_messenger->add_dispatcher_tail(&mgr_client);
- mgr_messenger->add_dispatcher_tail(this); // for auth ms_* calls
-
- 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 (int i = 0; i < PAXOS_NUM; ++i) {
- paxos_service[i]->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 {
- bufferlist::iterator p = bl.begin();
- ::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 (int i = 0; i < PAXOS_NUM; ++i) {
- paxos_service[i]->refresh(need_bootstrap);
- }
- for (int i = 0; i < PAXOS_NUM; ++i) {
- paxos_service[i]->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();
-
- state = STATE_SHUTDOWN;
-
- g_conf->remove_observer(this);
-
- if (admin_hook) {
- AdminSocket* admin_socket = cct->get_admin_socket();
- admin_socket->unregister_command("mon_status");
- admin_socket->unregister_command("quorum_status");
- admin_socket->unregister_command("sync_force");
- admin_socket->unregister_command("add_bootstrap_peer_hint");
- admin_socket->unregister_command("quorum enter");
- admin_socket->unregister_command("quorum exit");
- admin_socket->unregister_command("ops");
- admin_socket->unregister_command("sessions");
- 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 (vector<PaxosService*>::iterator p = paxos_service.begin(); p != paxos_service.end(); ++p)
- (*p)->shutdown();
- health_monitor->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();
-
- 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;
- }
-
- log_client.shutdown();
-
- // unlock before msgr shutdown...
- lock.Unlock();
-
- messenger->shutdown(); // last thing! ceph_mon.cc will delete mon.
- mgr_messenger->shutdown();
-}
-
-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::bootstrap()
-{
- dout(10) << "bootstrap" << dendl;
- wait_for_paxos_write();
-
- sync_reset_requester();
- unregister_cluster_logger();
- cancel_probe_timeout();
-
- // note my rank
- int newrank = monmap->get_rank(messenger->get_myaddr());
- 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 != 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)
- messenger->send_message(new MMonProbe(monmap->fsid, MMonProbe::OP_PROBE, name, has_ever_joined),
- monmap->get_inst(i));
- }
- for (set<entity_addr_t>::iterator p = extra_probe_peers.begin();
- p != extra_probe_peers.end();
- ++p) {
- if (*p != messenger->get_myaddr()) {
- entity_inst_t i;
- i.name = entity_name_t::MON(-1);
- i.addr = *p;
- messenger->send_message(new MMonProbe(monmap->fsid, MMonProbe::OP_PROBE, name, has_ever_joined), i);
- }
- }
-}
-
-bool Monitor::_add_bootstrap_peer_hint(string cmd, cmdmap_t& cmdmap, ostream& ss)
-{
- string addrstr;
- if (!cmd_getval(g_ceph_context, cmdmap, "addr", addrstr)) {
- ss << "unable to parse address string value '"
- << cmd_vartype_stringify(cmdmap["addr"]) << "'";
- return false;
- }
- dout(10) << "_add_bootstrap_peer_hint '" << cmd << "' '"
- << addrstr << "'" << dendl;
-
- entity_addr_t addr;
- const char *end = 0;
- if (!addr.parse(addrstr.c_str(), &end)) {
- ss << "failed to parse addr '" << addrstr << "'; syntax is 'add_bootstrap_peer_hint ip[:port]'";
- return false;
- }
-
- if (is_leader() || is_peon()) {
- ss << "mon already active; ignoring bootstrap hint";
- return true;
- }
-
- if (addr.get_port() == 0)
- addr.set_port(CEPH_MON_PORT);
-
- extra_probe_peers.insert(addr);
- ss << "adding peer " << addr << " to list: " << extra_probe_peers;
- return true;
-}
-
-// called by bootstrap(), or on leader|peon -> electing
-void Monitor::_reset()
-{
- dout(10) << __func__ << dendl;
-
- cancel_probe_timeout();
- timecheck_finish();
- health_events_cleanup();
- health_check_log_times.clear();
- scrub_event_cancel();
-
- leader_since = utime_t();
- if (!quorum.empty()) {
- exited_quorum = ceph_clock_now();
- }
- quorum.clear();
- outside_quorum.clear();
- quorum_feature_map.clear();
-
- scrub_reset();
-
- paxos->restart();
-
- for (vector<PaxosService*>::iterator p = paxos_service.begin(); p != paxos_service.end(); ++p)
- (*p)->restart();
- health_monitor->finish();
-}
-
-
-// -----------------------------------------------------------
-// sync
-
-set<string> Monitor::get_sync_targets_names()
-{
- set<string> targets;
- targets.insert(paxos->get_name());
- for (int i = 0; i < PAXOS_NUM; ++i)
- paxos_service[i]->get_store_prefixes(targets);
- ConfigKeyService *config_key_service_ptr = dynamic_cast<ConfigKeyService*>(config_key_service);
- assert(config_key_service_ptr);
- config_key_service_ptr->get_store_prefixes(targets);
- return targets;
-}
-
-
-void Monitor::sync_timeout()
-{
- dout(10) << __func__ << dendl;
- 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;
- assert(0 == "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;
- assert(0 == "error reading the store");
- }
- 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_inst_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_inst_t &other, bool full)
-{
- dout(10) << __func__ << " " << other << (full ? " full" : " recent") << dendl;
-
- 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 = 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);
-
- 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();
-
- 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 = other;
-
- 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_message(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);
- 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;
-
- 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);
- }
-
- 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);
-
- assert(g_conf->mon_sync_requester_kill_at != 9);
-
- init_paxos();
-
- 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;
- assert(0 == "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;
- }
-
- 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;
- 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.entity = m->get_source_inst();
- 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;
- }
-
- 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 left = g_conf->mon_sync_max_payload_size;
- while (sp.last_committed < paxos->get_version() && left > 0) {
- bufferlist bl;
- sp.last_committed++;
-
- int err = store->get(paxos->get_name(), sp.last_committed, bl);
- assert(err == 0);
-
- tx->put(paxos->get_name(), sp.last_committed, bl);
- left -= bl.length();
- dout(20) << __func__ << " including paxos state " << sp.last_committed
- << dendl;
- }
- reply->last_committed = sp.last_committed;
-
- if (sp.full && left > 0) {
- sp.synchronizer->get_chunk_tx(tx, 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;
-
- 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_inst() != 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();
-
- 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_message(r, sync_provider);
-
- 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_inst() != sync_provider) {
- dout(10) << __func__ << " source does not match, discarding" << dendl;
- return;
- }
-
- assert(state == STATE_SYNCHRONIZING);
- 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);
-
- 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.entity << 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;
- assert(is_probing() || is_synchronizing());
- 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__ << " 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 (missing) {
- dout(1) << " peer " << m->get_source_addr() << " missing features "
- << missing << dendl;
- if (m->get_connection()->has_feature(CEPH_FEATURE_OSD_PRIMARY_AFFINITY)) {
- MMonProbe *r = new MMonProbe(monmap->fsid, MMonProbe::OP_MISSING_FEATURES,
- name, has_ever_joined);
- 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);
- 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_addr()
- << " to list of hints" << dendl;
- extra_probe_peers.insert(m->get_source_addr());
- }
-
- 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 {
- dout(10) << " peer name is " << peer_name << dendl;
- }
-
- // new initial peer?
- if (monmap->get_epoch() == 0 &&
- monmap->contains(m->name) &&
- monmap->get_addr(m->name).is_blank_ip()) {
- dout(1) << " learned initial mon " << m->name << " addr " << m->get_source_addr() << dendl;
- monmap->set_addr(m->name, m->get_source_addr());
-
- bootstrap();
- return;
- }
-
- // end discover phase
- if (!is_probing()) {
- return;
- }
-
- assert(paxos != NULL);
-
- if (is_synchronizing()) {
- dout(10) << " currently syncing" << dendl;
- return;
- }
-
- entity_inst_t other = m->get_source_inst();
-
- 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;
- }
- }
-
- // 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_addr(name).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;
- messenger->send_message(new MMonJoin(monmap->fsid, name, messenger->get_myaddr()),
- monmap->get_inst(*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->size() / 2 + 1;
- 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 new 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);
- 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(),
- metadata);
-}
-
-const utime_t& Monitor::get_leader_since() const
-{
- assert(state == STATE_LEADER);
- return leader_since;
-}
-
-epoch_t Monitor::get_epoch()
-{
- return elector.get_epoch();
-}
-
-void Monitor::_finish_svc_election()
-{
- assert(state == STATE_LEADER || state == STATE_PEON);
-
- for (auto p : paxos_service) {
- // we already called election_finished() on monmon(); avoid callig twice
- if (state == STATE_LEADER && p == monmon())
- continue;
- p->election_finished();
- }
-}
-
-void Monitor::win_election(epoch_t epoch, set<int>& active, uint64_t features,
- const mon_feature_t& mon_features,
- const map<int,Metadata>& metadata)
-{
- dout(10) << __func__ << " epoch " << epoch << " quorum " << active
- << " features " << features
- << " mon_features " << mon_features
- << dendl;
- assert(is_electing());
- state = STATE_LEADER;
- leader_since = ceph_clock_now();
- leader = rank;
- quorum = active;
- quorum_con_features = features;
- quorum_mon_features = mon_features;
- pending_metadata = metadata;
- outside_quorum.clear();
-
- clog->info() << "mon." << name << "@" << rank
- << " won leader election with quorum " << 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();
- health_monitor->start(epoch);
-
- 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();
- 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)
-{
- state = STATE_PEON;
- leader_since = utime_t();
- leader = l;
- quorum = q;
- outside_quorum.clear();
- quorum_con_features = features;
- quorum_mon_features = mon_features;
- dout(10) << "lose_election, epoch " << epoch << " leader is mon" << leader
- << " quorum is " << quorum << " features are " << quorum_con_features
- << " mon_features are " << quorum_mon_features
- << dendl;
-
- paxos->peon_init();
- _finish_svc_election();
- health_monitor->start(epoch);
-
- logger->inc(l_mon_election_lose);
-
- finish_election();
-
- if ((quorum_con_features & CEPH_FEATURE_MON_METADATA) &&
- !HAVE_FEATURE(quorum_con_features, SERVER_LUMINOUS)) {
- // for pre-luminous mons only
- Metadata sys_info;
- collect_metadata(&sys_info);
- messenger->send_message(new MMonMetadata(sys_info),
- monmap->get_inst(get_leader()));
- }
-}
-
-void Monitor::collect_metadata(Metadata *m)
-{
- collect_sys_info(m, g_ceph_context);
- (*m)["addr"] = stringify(messenger->get_myaddr());
-}
-
-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();
-
- // am i named properly?
- string cur_name = monmap->get_name(messenger->get_myaddr());
- if (cur_name != name) {
- dout(10) << " renaming myself from " << cur_name << " -> " << name << dendl;
- messenger->send_message(new MMonJoin(monmap->fsid, name, messenger->get_myaddr()),
- monmap->get_inst(*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);
- if (quorum_con_features & CEPH_FEATURE_OSD_ERASURE_CODES) {
- 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);
- }
- if (quorum_con_features & CEPH_FEATURE_ERASURE_CODE_PLUGINS_V2) {
- new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_ERASURE_CODE_PLUGINS_V2);
- }
- if (quorum_con_features & CEPH_FEATURE_ERASURE_CODE_PLUGINS_V3) {
- 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)) {
- 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.
- 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)) {
- 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.
- assert(HAVE_FEATURE(quorum_con_features, SERVER_LUMINOUS));
- new_features.incompat.insert(CEPH_MON_FEATURE_INCOMPAT_LUMINOUS);
- }
-
- 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_OSD_ERASURE_CODES)) {
- required_features |= CEPH_FEATURE_OSD_ERASURE_CODES;
- }
- if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_OSDMAP_ENC)) {
- required_features |= CEPH_FEATURE_OSDMAP_ENC;
- }
- if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_ERASURE_CODE_PLUGINS_V2)) {
- required_features |= CEPH_FEATURE_ERASURE_CODE_PLUGINS_V2;
- }
- if (features.incompat.contains(CEPH_MON_FEATURE_INCOMPAT_ERASURE_CODE_PLUGINS_V3)) {
- required_features |= CEPH_FEATURE_ERASURE_CODE_PLUGINS_V3;
- }
- 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;
- }
-
- // 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;
- }
- 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()));
-
- 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
-
- 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_addr_t>::iterator p = extra_probe_peers.begin();
- p != extra_probe_peers.end();
- ++p)
- f->dump_stream("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_stream("entity") << p->second.entity;
- 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;
- do_health_to_clog();
- 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();
- } 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;
-
- if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- 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;
- clog->health(level) << "overall " << summary;
- health_status_cache.summary = summary;
- health_status_cache.overall = level;
- } else {
- // for jewel only
- list<string> status;
- health_status_t overall = get_health(status, NULL, NULL);
- dout(25) << __func__
- << (force ? " (force)" : "")
- << dendl;
-
- string summary = joinify(status.begin(), status.end(), string("; "));
-
- if (!force &&
- overall == health_status_cache.overall &&
- !health_status_cache.summary.empty() &&
- health_status_cache.summary == summary) {
- // we got a dup!
- return;
- }
-
- clog->info() << summary;
-
- health_status_cache.overall = overall;
- health_status_cache.summary = summary;
- }
-}
-
-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;
- bool compat = g_conf->mon_health_preluminous_compat;
- bool compat_warn = g_conf->get_val<bool>("mon_health_preluminous_compat_warning");
- 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;
- } else {
- // one-liner: HEALTH_FOO[ thing1[; thing2 ...]]
- *plain = stringify(r);
- if (summary.size()) {
- *plain += sep1;
- *plain += summary;
- }
- *plain += "\n";
- }
-
- const std::string old_fields_message = "'ceph health' JSON format has "
- "changed in luminous. If you see this your monitoring system is "
- "scraping the wrong fields. Disable this with 'mon health preluminous "
- "compat warning = false'";
-
- if (f && (compat || compat_warn)) {
- health_status_t cr = compat_warn ? min(HEALTH_WARN, r) : r;
- f->open_array_section("summary");
- if (compat_warn) {
- f->open_object_section("item");
- f->dump_stream("severity") << HEALTH_WARN;
- f->dump_string("summary", old_fields_message);
- f->close_section();
- }
- if (compat) {
- for (auto& svc : paxos_service) {
- svc->get_health_checks().dump_summary_compat(f);
- }
- }
- f->close_section();
- f->dump_stream("overall_status") << cr;
- }
-
- if (want_detail) {
- if (f && (compat || compat_warn)) {
- f->open_array_section("detail");
- if (compat_warn) {
- f->dump_string("item", old_fields_message);
- }
- }
-
- for (auto& svc : paxos_service) {
- svc->get_health_checks().dump_detail(f, plain, compat);
- }
-
- if (f && (compat || compat_warn)) {
- f->close_section();
- }
- }
- if (f) {
- f->close_section();
- }
- 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";
- }
- }
-}
-
-health_status_t Monitor::get_health(list<string>& status,
- bufferlist *detailbl,
- Formatter *f)
-{
- list<pair<health_status_t,string> > summary;
- list<pair<health_status_t,string> > detail;
-
- if (f)
- f->open_object_section("health");
-
- for (vector<PaxosService*>::iterator p = paxos_service.begin();
- p != paxos_service.end();
- ++p) {
- PaxosService *s = *p;
- s->get_health(summary, detailbl ? &detail : NULL, cct);
- }
-
- health_monitor->get_health(summary, (detailbl ? &detail : NULL));
-
- health_status_t overall = HEALTH_OK;
- if (!timecheck_skews.empty()) {
- list<string> warns;
- for (map<entity_inst_t,double>::iterator i = timecheck_skews.begin();
- i != timecheck_skews.end(); ++i) {
- entity_inst_t inst = i->first;
- double skew = i->second;
- double latency = timecheck_latencies[inst];
- string name = monmap->get_name(inst.addr);
- ostringstream tcss;
- health_status_t tcstatus = timecheck_status(tcss, skew, latency);
- if (tcstatus != HEALTH_OK) {
- if (overall > tcstatus)
- overall = tcstatus;
- warns.push_back(name);
- ostringstream tmp_ss;
- tmp_ss << "mon." << name
- << " addr " << inst.addr << " " << tcss.str()
- << " (latency " << latency << "s)";
- detail.push_back(make_pair(tcstatus, tmp_ss.str()));
- }
- }
- if (!warns.empty()) {
- ostringstream ss;
- ss << "clock skew detected on";
- while (!warns.empty()) {
- ss << " mon." << warns.front();
- warns.pop_front();
- if (!warns.empty())
- ss << ",";
- }
- status.push_back(ss.str());
- summary.push_back(make_pair(HEALTH_WARN, "Monitor clock skew detected "));
- }
- }
-
- if (f)
- f->open_array_section("summary");
- if (!summary.empty()) {
- while (!summary.empty()) {
- if (overall > summary.front().first)
- overall = summary.front().first;
- status.push_back(summary.front().second);
- if (f) {
- f->open_object_section("item");
- f->dump_stream("severity") << summary.front().first;
- f->dump_string("summary", summary.front().second);
- f->close_section();
- }
- summary.pop_front();
- }
- }
- if (f)
- f->close_section();
-
- stringstream fss;
- fss << overall;
- status.push_front(fss.str());
- if (f)
- f->dump_stream("overall_status") << overall;
-
- if (f)
- f->open_array_section("detail");
- while (!detail.empty()) {
- if (f)
- f->dump_string("item", detail.front().second);
- else if (detailbl != NULL) {
- detailbl->append(detail.front().second);
- detailbl->append('\n');
- }
- detail.pop_front();
- }
- if (f)
- f->close_section();
-
- if (f)
- f->close_section();
-
- return overall;
-}
-
-void Monitor::get_cluster_status(stringstream &ss, Formatter *f)
-{
- if (f)
- f->open_object_section("status");
-
- if (f) {
- f->dump_stream("fsid") << monmap->get_fsid();
- if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- get_health_status(false, f, nullptr);
- } else {
- list<string> health_str;
- get_health(health_str, nullptr, f);
- }
- 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->open_object_section("monmap");
- monmap->dump(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");
- pgservice->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->close_section();
- } else {
- ss << " cluster:\n";
- ss << " id: " << monmap->get_fsid() << "\n";
-
- string health;
- if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- get_health_status(false, nullptr, &health,
- "\n ", "\n ");
- } else {
- list<string> ls;
- get_health(ls, NULL, f);
- health = joinify(ls.begin(), ls.end(),
- string("\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;
- 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()->get_fsmap().filesystem_count() > 0) {
- 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) {
- ss << " " << p.first << ": " << string(maxlen - p.first.size(), ' ')
- << p.second.get_summary() << "\n";
- }
- }
-
- ss << "\n \n data:\n";
- pgservice->print_summary(NULL, &ss);
- ss << "\n ";
- }
-}
-
-void Monitor::_generate_command_map(map<string,cmd_vartype>& cmdmap,
- map<string,string> ¶m_str_map)
-{
- for (map<string,cmd_vartype>::const_iterator 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, string &module, string &prefix,
- const map<string,cmd_vartype>& 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,
- CEPH_ENTITY_TYPE_MON,
- s->entity_name,
- module, prefix, param_str_map,
- cmd_r, cmd_w, cmd_x);
-
- dout(10) << __func__ << " " << (capable ? "" : "not ") << "capable" << dendl;
- return capable;
-}
-
-void Monitor::format_command_descriptions(const std::vector<MonCommand> &commands,
- Formatter *f,
- bufferlist *rdata,
- bool hide_mgr_flag)
-{
- int cmdnum = 0;
- f->open_object_section("command_descriptions");
- for (const auto &cmd : commands) {
- unsigned flags = cmd.flags;
- if (hide_mgr_flag) {
- flags &= ~MonCommand::FLAG_MGR;
- }
- ostringstream secname;
- secname << "cmd" << setfill('0') << std::setw(3) << cmdnum;
- dump_cmddesc_to_json(f, secname.str(),
- cmd.cmdstring, cmd.helpstring, cmd.module,
- cmd.req_perms, cmd.availability, flags);
- cmdnum++;
- }
- f->close_section(); // command_descriptions
-
- f->flush(*rdata);
-}
-
-bool Monitor::is_keyring_required()
-{
- string auth_cluster_required = g_conf->auth_supported.empty() ?
- g_conf->auth_cluster_required : g_conf->auth_supported;
- string auth_service_required = g_conf->auth_supported.empty() ?
- g_conf->auth_service_required : g_conf->auth_supported;
-
- return auth_service_required == "cephx" ||
- auth_cluster_required == "cephx";
-}
-
-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) {
- Mutex::Locker l(mon->lock);
- mon->mgr_proxy_bytes -= size;
- mon->reply_command(op, r, outs, outbl, 0);
- }
-};
-
-void Monitor::handle_command(MonOpRequestRef op)
-{
- 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 = static_cast<MonSession *>(
- m->get_connection()->get_priv());
- if (!session) {
- dout(5) << __func__ << " dropping stray message " << *m << dendl;
- return;
- }
- BOOST_SCOPE_EXIT_ALL(=) {
- session->put();
- };
-
- if (m->cmd.empty()) {
- string rs = "No command supplied";
- reply_command(op, -EINVAL, rs, 0);
- return;
- }
-
- string prefix;
- vector<string> fullcmd;
- map<string, cmd_vartype> 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");
- // hide mgr commands until luminous upgrade is complete
- bool hide_mgr_flag =
- osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS;
-
- std::vector<MonCommand> commands;
-
- // only include mgr commands once all mons are upgrade (and we've dropped
- // the hard-coded PGMonitor commands)
- if (quorum_mon_features.contains_all(ceph::features::mon::FEATURE_LUMINOUS)) {
- commands = static_cast<MgrMonitor*>(
- paxos_service[PAXOS_MGR])->get_command_descs();
- }
-
- for (auto& c : leader_mon_commands) {
- commands.push_back(c);
- }
-
- format_command_descriptions(commands, f, &rdata, hide_mgr_flag);
- 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;
- (cmd_is_rw ? audit_clog->info() : audit_clog->debug())
- << "from='" << session->inst << "' "
- << "entity='" << session->entity_name << "' "
- << "cmd=" << m->cmd << ": access denied";
- reply_command(op, -EACCES, "access denied", 0);
- return;
- }
-
- (cmd_is_rw ? audit_clog->info() : audit_clog->debug())
- << "from='" << session->inst << "' "
- << "entity='" << session->entity_name << "' "
- << "cmd=" << m->cmd << ": dispatch";
-
- if (mon_cmd->is_mgr() &&
- osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- 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<uint64_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 != "osd last-stat-seq") {
- osdmon()->dispatch(op);
- return;
- }
-
- if (module == "pg") {
- pgmon()->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") {
- 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;
- utime_t start = ceph_clock_now();
- store->compact();
- utime_t end = ceph_clock_now();
- end -= start;
- dout(1) << "finished manual compaction in " << end << " seconds" << dendl;
- ostringstream oss;
- oss << "compacted " << g_conf->get_val<std::string>("mon_keyvaluedb") << " in " << end << " 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) {
- entity_inst_t inst = i.first;
- double skew = i.second;
- double latency = timecheck_latencies[inst];
- string name = monmap->get_name(inst.addr);
- 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, true, &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") {
- if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- string plain;
- get_health_status(detail == "detail", f.get(), f ? nullptr : &plain);
- if (f) {
- f->flush(rdata);
- } else {
- rdata.append(plain);
- }
- } else {
- list<string> health_str;
- get_health(health_str, detail == "detail" ? &rdata : NULL, f.get());
- if (f) {
- f->flush(ds);
- ds << '\n';
- } else {
- assert(!health_str.empty());
- ds << health_str.front();
- health_str.pop_front();
- if (!health_str.empty()) {
- ds << ' ';
- ds << joinify(health_str.begin(), health_str.end(), string("; "));
- }
- }
- bufferlist comb;
- comb.append(ds);
- if (detail == "detail")
- comb.append(rdata);
- rdata = comb;
- }
- } else if (prefix == "df") {
- bool verbose = (detail == "detail");
- if (f)
- f->open_object_section("stats");
-
- pgservice->dump_fs_stats(&ds, f.get(), verbose);
- if (!f)
- ds << '\n';
- pgservice->dump_pool_stats(osdmon()->osdmap, &ds, f.get(), verbose);
-
- if (f) {
- f->close_section();
- f->flush(ds);
- ds << '\n';
- }
- } else {
- assert(0 == "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);
-
- if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- get_health_status(true, f.get(), nullptr);
- } else {
- list<string> health_str;
- get_health(health_str, nullptr, f.get());
- }
-
- monmon()->dump_info(f.get());
- osdmon()->dump_info(f.get());
- mdsmon()->dump_info(f.get());
- authmon()->dump_info(f.get());
- pgservice->dump_info(f.get());
-
- paxos->dump_info(f.get());
-
- f->close_section();
- f->flush(rdata);
-
- ostringstream ss2;
- ss2 << "report " << rdata.crc32c(CEPH_MON_PORT);
- 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());
- 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());
- }
- 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_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") {
- string validate1, validate2;
- cmd_getval(g_ceph_context, cmdmap, "validate1", validate1);
- cmd_getval(g_ceph_context, cmdmap, "validate2", validate2);
- if (validate1 != "--yes-i-really-mean-it" ||
- validate2 != "--i-know-what-i-am-doing") {
- 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);
- 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) {
- f->open_object_section(p.first.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;
- }
-
- 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());
- 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_addr() != messenger->get_myaddr()) {
- 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->client_inst = req->get_source_inst();
- 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);
- }
- messenger->send_message(forward, monmap->get_inst(mon));
- op->mark_forwarded();
- 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 {
- explicit AnonConnection(CephContext *cct) : Connection(cct, NULL) {}
-
- int send_message(Message *m) override {
- assert(!"send_message on anonymous connection");
- }
- void send_keepalive() override {
- assert(!"send_keepalive on anonymous connection");
- }
- void mark_down() override {
- // silently ignore
- }
- void mark_disposable() override {
- // silengtly ignore
- }
- bool is_connected() override { return false; }
-};
-
-//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 " << m->client
- << " via " << m->get_source_inst() << dendl;
- MonSession *session = op->get_session();
- 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();
- assert(req != NULL);
-
- ConnectionRef c(new AnonConnection(cct));
- MonSession *s = new MonSession(req->get_source_inst(),
- static_cast<Connection*>(c.get()));
- c->set_priv(s->get());
- c->set_peer_addr(m->client.addr);
- c->set_peer_type(m->client.name.type());
- c->set_features(m->con_features);
-
- 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();
-
- /* Because this is a special fake connection, we need to break
- the ref loop between Connection and MonSession differently
- than we normally do. Here, the Message refers to the Connection
- which refers to the Session, and nobody else refers to the Connection
- or the Session. And due to the special nature of this message,
- nobody refers to the Connection via the Session. So, clear out that
- half of the ref loop.*/
- s->con.reset(NULL);
-
- dout(10) << " mesg " << req << " from " << m->get_source_addr() << dendl;
-
- _ms_dispatch(req);
- s->put();
- }
-}
-
-void Monitor::try_send_message(Message *m, const entity_inst_t& to)
-{
- dout(10) << "try_send_message " << *m << " to " << to << dendl;
-
- bufferlist bl;
- encode_message(m, quorum_con_features, bl);
-
- messenger->send_message(m, to);
-
- for (int i=0; i<(int)monmap->size(); i++) {
- if (i != rank)
- messenger->send_message(new MRoute(bl, to), monmap->get_inst(i));
- }
-}
-
-void Monitor::send_reply(MonOpRequestRef op, Message *reply)
-{
- op->mark_event(__func__);
-
- MonSession *session = op->get_session();
- 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 " << *m->msg << " to " << m->dest << dendl;
- else
- dout(10) << "handle_route null to " << m->dest << 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());
- }
- 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, trying to send anyway" << dendl;
- if (m->msg) {
- messenger->send_message(m->msg, m->dest);
- m->msg = NULL;
- }
- }
-}
-
-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) {
- assert(rr->session->routed_request_tids.count(p->first));
- rr->session->routed_request_tids.erase(p->first);
- }
- delete rr;
- } else {
- bufferlist::iterator q = rr->request_bl.begin();
- 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 = rr->client_inst;
- forward->set_priority(req->get_priority());
- messenger->send_message(forward, monmap->get_inst(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->inst
- << " features 0x" << std::hex << s->con_features << std::dec << dendl;
- assert(s->con);
- assert(!s->closed);
- for (set<uint64_t>::iterator p = s->routed_request_tids.begin();
- p != s->routed_request_tids.end();
- ++p) {
- 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(NULL);
- 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()
-{
- Mutex::Locker l(session_map_lock);
- while (!session_map.sessions.empty()) {
- MonSession *s = session_map.sessions.front();
- remove_session(s);
- if (logger)
- logger->inc(l_mon_session_rm);
- }
- if (logger)
- logger->set(l_mon_num_sessions, session_map.get_size());
-}
-
-void Monitor::send_command(const entity_inst_t& inst,
- const vector<string>& com)
-{
- dout(10) << "send_command " << inst << "" << com << dendl;
- MMonCommand *c = new MMonCommand(monmap->fsid);
- c->cmd = com;
- try_send_message(c, inst);
-}
-
-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) {
- Mutex::Locker 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.
- remove_session(s);
- }
- s->put();
- 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();
- {
- Mutex::Locker l(session_map_lock);
- s = session_map.new_session(m->get_source_inst(), con.get());
- }
- assert(s);
- con->set_priv(s->get());
- 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->put();
- } else {
- dout(20) << __func__ << " existing session " << s << " for " << s->inst
- << dendl;
- }
-
- 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();
- }
- dout(20) << " caps " << s->caps.get_str() << dendl;
-
- if ((is_synchronizing() ||
- (s->global_id == 0 && !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();
- 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 */
- bool dealt_with = true;
- 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);
- break;
-
- case CEPH_MSG_PING:
- handle_ping(op);
- break;
-
- /* MMonGetMap may be used by clients to obtain a monmap *before*
- * authenticating with the monitor. We need to handle these without
- * checking caps because, even on a cluster without cephx, we only set
- * session caps *after* the auth handshake. A good example of this
- * is when a client calls MonClient::get_monmap_privately(), which does
- * not authenticate when obtaining a monmap.
- */
- case CEPH_MSG_MON_GET_MAP:
- handle_mon_get_map(op);
- break;
-
- case CEPH_MSG_MON_METADATA:
- return handle_mon_metadata(op);
-
- default:
- dealt_with = false;
- break;
- }
- if (dealt_with)
- return;
-
- /* well, maybe the op belongs to a service... */
- op->set_type_service();
- /* deal with all messages which caps should be checked somewhere else */
- dealt_with = true;
- 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:
- paxos_service[PAXOS_OSDMAP]->dispatch(op);
- break;
-
- // MDSs
- case MSG_MDS_BEACON:
- case MSG_MDS_OFFLOAD_TARGETS:
- paxos_service[PAXOS_MDSMAP]->dispatch(op);
- break;
-
- // Mgrs
- case MSG_MGR_BEACON:
- paxos_service[PAXOS_MGR]->dispatch(op);
- break;
-
- // MgrStat
- case CEPH_MSG_STATFS:
- // this is an ugly hack, sorry! force the version to 1 so that we do
- // not run afoul of the is_readable() paxos check. the client is going
- // by the pgmonitor version and the MgrStatMonitor version will lag behind
- // that until we complete the upgrade. The paxos ordering crap really
- // doesn't matter for statfs results, so just kludge around it here.
- if (osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
- ((MStatfs*)op->get_req())->version = 1;
- }
- case MSG_MON_MGR_REPORT:
- case MSG_GETPOOLSTATS:
- paxos_service[PAXOS_MGRSTAT]->dispatch(op);
- break;
-
- // pg
- case MSG_PGSTATS:
- paxos_service[PAXOS_PGMAP]->dispatch(op);
- break;
-
- // log
- case MSG_LOG:
- paxos_service[PAXOS_LOG]->dispatch(op);
- break;
-
- // handle_command() does its own caps checking
- case MSG_MON_COMMAND:
- op->set_type_command();
- handle_command(op);
- break;
-
- default:
- dealt_with = false;
- break;
- }
- if (dealt_with)
- 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;
- goto drop;
- }
-
- dealt_with = true;
- switch (op->get_req()->get_type()) {
-
- // misc
- case CEPH_MSG_MON_GET_VERSION:
- handle_get_version(op);
- break;
-
- case CEPH_MSG_MON_SUBSCRIBE:
- /* FIXME: check what's being subscribed, filter accordingly */
- handle_subscribe(op);
- break;
-
- default:
- dealt_with = false;
- break;
- }
- if (dealt_with)
- 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;
- goto drop;
- }
-
- /* messages that should only be sent by another monitor */
- dealt_with = true;
- switch (op->get_req()->get_type()) {
-
- case MSG_ROUTE:
- handle_route(op);
- break;
-
- case MSG_MON_PROBE:
- handle_probe(op);
- break;
-
- // Sync (i.e., the new slurp, but on steroids)
- case MSG_MON_SYNC:
- handle_sync(op);
- break;
- case MSG_MON_SCRUB:
- handle_scrub(op);
- break;
-
- /* 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());
- break;
-
- // monmap
- case MSG_MON_JOIN:
- op->set_type_service();
- paxos_service[PAXOS_MONMAP]->dispatch(op);
- break;
-
- // 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!
- break;
- }
-
- 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;
- break;
- }
-
- // sanitize
- if (pm->epoch > get_epoch()) {
- bootstrap();
- break;
- }
- if (pm->epoch != get_epoch()) {
- break;
- }
-
- paxos->dispatch(op);
- }
- break;
-
- // 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;
- break;
- }
- if (!is_probing() && !is_synchronizing()) {
- elector.dispatch(op);
- }
- break;
-
- case MSG_FORWARD:
- handle_forward(op);
- break;
-
- case MSG_TIMECHECK:
- handle_timecheck(op);
- break;
-
- case MSG_MON_HEALTH:
- health_monitor->dispatch(op);
- break;
-
- case MSG_MON_HEALTH_CHECKS:
- op->set_type_service();
- paxos_service[PAXOS_HEALTH]->dispatch(op);
- break;
-
- default:
- dealt_with = false;
- break;
- }
- if (!dealt_with) {
- dout(1) << "dropping unexpected " << *(op->get_req()) << dendl;
- goto drop;
- }
- return;
-
-drop:
- return;
-}
-
-void Monitor::handle_ping(MonOpRequestRef op)
-{
- MPing *m = static_cast<MPing*>(op->get_req());
- dout(10) << __func__ << " " << *m << dendl;
- MPing *reply = new MPing;
- entity_inst_t inst = m->get_source_inst();
- bufferlist payload;
- boost::scoped_ptr<Formatter> f(new JSONFormatter(true));
- f->open_object_section("pong");
-
- if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- get_health_status(false, f.get(), nullptr);
- } else {
- list<string> health_str;
- get_health(health_str, nullptr, f.get());
- }
-
- {
- 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;
- messenger->send_message(reply, inst);
-}
-
-void Monitor::timecheck_start()
-{
- dout(10) << __func__ << dendl;
- timecheck_cleanup();
- 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;
- assert(is_leader());
-
- if (monmap->size() == 1) {
- assert(0 == "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();
- }
- }
-
- 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;
- assert(timecheck_round % 2);
- timecheck_round ++;
- timecheck_round_start = utime_t();
-
- if (success) {
- assert(timecheck_waiting.empty());
- assert(timecheck_acks == quorum.size());
- timecheck_report();
- timecheck_check_skews();
- return;
- }
-
- dout(10) << __func__ << " " << timecheck_waiting.size()
- << " peers still waiting:";
- for (map<entity_inst_t,utime_t>::iterator p = timecheck_waiting.begin();
- p != timecheck_waiting.end(); ++p) {
- *_dout << " " << p->first.name;
- }
- *_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;
- assert(is_leader());
- assert((timecheck_round % 2) == 0);
- if (monmap->size() == 1) {
- assert(0 == "We are alone; we shouldn't have gotten here!");
- return;
- }
- assert(timecheck_latencies.size() == timecheck_skews.size());
-
- bool found_skew = false;
- for (map<entity_inst_t, double>::iterator p = timecheck_skews.begin();
- p != timecheck_skews.end(); ++p) {
-
- 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;
- assert(is_leader());
- assert((timecheck_round % 2) == 0);
- if (monmap->size() == 1) {
- assert(0 == "We are alone; we shouldn't have gotten here!");
- return;
- }
-
- 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;
-
- MTimeCheck *m = new MTimeCheck(MTimeCheck::OP_REPORT);
- m->epoch = get_epoch();
- m->round = timecheck_round;
-
- for (map<entity_inst_t, double>::iterator it = timecheck_skews.begin();
- it != timecheck_skews.end(); ++it) {
- 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__ << " " << it->first
- << " latency " << latency
- << " skew " << skew << dendl;
- }
- }
- do_output = false;
- entity_inst_t inst = monmap->get_inst(*q);
- dout(10) << __func__ << " send report to " << inst << dendl;
- messenger->send_message(m, inst);
- }
-}
-
-void Monitor::timecheck()
-{
- dout(10) << __func__ << dendl;
- assert(is_leader());
- if (monmap->size() == 1) {
- assert(0 == "We are alone; we shouldn't have gotten here!");
- return;
- }
- 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[messenger->get_myinst()] = 0.0;
- timecheck_latencies[messenger->get_myinst()] = 0.0;
-
- for (set<int>::iterator it = quorum.begin(); it != quorum.end(); ++it) {
- if (monmap->get_name(*it) == name)
- continue;
-
- entity_inst_t inst = monmap->get_inst(*it);
- utime_t curr_time = ceph_clock_now();
- timecheck_waiting[inst] = curr_time;
- MTimeCheck *m = new MTimeCheck(MTimeCheck::OP_PING);
- m->epoch = get_epoch();
- m->round = timecheck_round;
- dout(10) << __func__ << " send " << *m << " to " << inst << dendl;
- messenger->send_message(m, inst);
- }
-}
-
-health_status_t Monitor::timecheck_status(ostringstream &ss,
- const double skew_bound,
- const double latency)
-{
- health_status_t status = HEALTH_OK;
- 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)
-{
- MTimeCheck *m = static_cast<MTimeCheck*>(op->get_req());
- dout(10) << __func__ << " " << *m << dendl;
- /* handles PONG's */
- assert(m->op == MTimeCheck::OP_PONG);
-
- entity_inst_t other = m->get_source_inst();
- if (m->epoch < get_epoch()) {
- dout(1) << __func__ << " got old timecheck epoch " << m->epoch
- << " from " << other
- << " curr " << get_epoch()
- << " -- severely lagged? discard" << dendl;
- return;
- }
- 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();
-
- 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.
- */
- 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);
- 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;
- assert(timecheck_skews.size() == timecheck_acks);
- assert(timecheck_waiting.empty());
- // everyone has acked, so bump the round to finish it.
- timecheck_finish_round();
- }
-}
-
-void Monitor::handle_timecheck_peon(MonOpRequestRef op)
-{
- MTimeCheck *m = static_cast<MTimeCheck*>(op->get_req());
- dout(10) << __func__ << " " << *m << dendl;
-
- assert(is_peon());
- assert(m->op == MTimeCheck::OP_PING || m->op == MTimeCheck::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 == MTimeCheck::OP_REPORT) {
- assert((timecheck_round % 2) == 0);
- timecheck_latencies.swap(m->latencies);
- timecheck_skews.swap(m->skews);
- return;
- }
-
- assert((timecheck_round % 2) != 0);
- MTimeCheck *reply = new MTimeCheck(MTimeCheck::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)
-{
- MTimeCheck *m = static_cast<MTimeCheck*>(op->get_req());
- dout(10) << __func__ << " " << *m << dendl;
-
- if (is_leader()) {
- if (m->op != MTimeCheck::OP_PONG) {
- dout(1) << __func__ << " drop unexpected msg (not pong)" << dendl;
- } else {
- handle_timecheck_leader(op);
- }
- } else if (is_peon()) {
- if (m->op != MTimeCheck::OP_PING && m->op != MTimeCheck::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();
- assert(s);
-
- for (map<string,ceph_mon_subscribe_item>::iterator p = m->what.begin();
- p != m->what.end();
- ++p) {
- // 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) {
- Mutex::Locker l(session_map_lock);
- session_map.remove_sub((it++)->second);
- } else {
- ++it;
- }
- }
- }
-
- {
- Mutex::Locker 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];
- 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)) {
- if (monmap->get_required_features().contains_all(
- ceph::features::mon::FEATURE_LUMINOUS)) {
- osdmon()->check_pg_creates_sub(s->sub_map["osd_pg_creates"]);
- } else {
- pgmon()->check_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]);
- }
- }
-
- 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();
- 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;
-
- MonSession *s = static_cast<MonSession *>(con->get_priv());
- if (!s)
- return false;
-
- // break any con <-> session ref cycle
- s->con->set_priv(NULL);
-
- if (is_shutdown())
- return false;
-
- Mutex::Locker l(lock);
-
- dout(10) << "reset/close on session " << s->inst << dendl;
- if (!s->closed) {
- Mutex::Locker l(session_map_lock);
- remove_session(s);
- }
- s->put();
- 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;
- bufferlist::iterator it = bl.begin();
- ::decode(mon_metadata, it);
-
- pending_metadata = mon_metadata;
- return 0;
-}
-
-int Monitor::get_mon_metadata(int mon, Formatter *f, ostream& err)
-{
- 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<int> > 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(it->first);
- }
-
- dump_services(f, mons, "mon");
- return 0;
-}
-
-// ----------------------------------------------
-// scrub
-
-int Monitor::scrub_start()
-{
- dout(10) << __func__ << dendl;
- 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()
-{
- assert(is_leader());
- 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;
- messenger->send_message(r, monmap->get_inst(*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) {
- 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();
- 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)
-{
- assert(r != NULL);
- assert(start != NULL);
- 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);
- 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 (vector<PaxosService*>::iterator p = paxos_service.begin(); p != paxos_service.end(); ++p) {
- (*p)->tick();
- (*p)->maybe_trim();
- }
-
- // trim sessions
- {
- Mutex::Locker 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->inst.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->inst
- << " (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->inst
- << " 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();
- }
-
- new_tick();
-}
-
-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;
- 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 {
- bufferlist::iterator i = bl.begin();
- 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, 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);
- }
-}
-
-bool Monitor::ms_get_authorizer(int service_id, AuthAuthorizer **authorizer,
- bool force_new)
-{
- 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, nullptr);
- 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, info,
- secret, (uint64_t)-1);
- 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, 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);
-
- bufferlist::iterator iter = ticket_data.begin();
- CephXTicketHandler handler(g_ceph_context, service_id);
- ::decode(handler.ticket, iter);
-
- handler.session_key = info.session_key;
-
- *authorizer = handler.build_authorizer(0);
-
- return true;
-}
-
-bool Monitor::ms_verify_authorizer(Connection *con, int peer_type,
- int protocol, bufferlist& authorizer_data,
- bufferlist& authorizer_reply,
- bool& isvalid, CryptoKey& session_key)
-{
- dout(10) << "ms_verify_authorizer " << con->get_peer_addr()
- << " " << ceph_entity_type_name(peer_type)
- << " protocol " << protocol << dendl;
-
- if (is_shutdown())
- return false;
-
- if (peer_type == CEPH_ENTITY_TYPE_MON &&
- auth_cluster_required.is_supported_auth(CEPH_AUTH_CEPHX)) {
- // monitor, and cephx is enabled
- isvalid = false;
- if (protocol == CEPH_AUTH_CEPHX) {
- bufferlist::iterator iter = authorizer_data.begin();
- CephXServiceTicketInfo auth_ticket_info;
-
- if (authorizer_data.length()) {
- bool ret = cephx_verify_authorizer(g_ceph_context, &keyring, iter,
- auth_ticket_info, authorizer_reply);
- if (ret) {
- session_key = auth_ticket_info.session_key;
- isvalid = true;
- } else {
- dout(0) << "ms_verify_authorizer bad authorizer from mon " << con->get_peer_addr() << dendl;
- }
- }
- } else {
- dout(0) << "ms_verify_authorizer cephx enabled, but no authorizer (required for mon)" << dendl;
- }
- } else {
- // who cares.
- isvalid = true;
- }
- return true;
-}
Code Review / stor4nfv.git / blobdiff