update source to 12.2.11

[ceph.git] / ceph / src / mon / Monitor.cc

// -*- 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", NULL, 0, unit_t(BYTES));
    pcb.add_u64(l_cluster_osd_bytes_used, "osd_bytes_used", "Used space", NULL, 0, unit_t(BYTES));
    pcb.add_u64(l_cluster_osd_bytes_avail, "osd_bytes_avail", "Available space", NULL, 0, unit_t(BYTES));
    pcb.add_u64(l_cluster_num_pool, "num_pool", "Pools");
    pcb.add_u64(l_cluster_num_pg, "num_pg", "Placement groups");
    pcb.add_u64(l_cluster_num_pg_active_clean, "num_pg_active_clean", "Placement groups in active+clean state");
    pcb.add_u64(l_cluster_num_pg_active, "num_pg_active", "Placement groups in active state");
    pcb.add_u64(l_cluster_num_pg_peering, "num_pg_peering", "Placement groups in peering state");
    pcb.add_u64(l_cluster_num_object, "num_object", "Objects");
    pcb.add_u64(l_cluster_num_object_degraded, "num_object_degraded", "Degraded (missing replicas) objects");
    pcb.add_u64(l_cluster_num_object_misplaced, "num_object_misplaced", "Misplaced (wrong location in the cluster) objects");
    pcb.add_u64(l_cluster_num_object_unfound, "num_object_unfound", "Unfound objects");
    pcb.add_u64(l_cluster_num_bytes, "num_bytes", "Size of all objects");
    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;

  lock.Unlock();
  g_conf->remove_observer(this);
  lock.Lock();

  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();

  log_client.shutdown();

  // unlock before msgr shutdown...
  lock.Unlock();

  // shutdown messenger before removing logger from perfcounter collection, 
  // otherwise _ms_dispatch() will try to update deleted logger
  messenger->shutdown();
  mgr_messenger->shutdown();

  if (logger) {
    cct->get_perfcounters_collection()->remove(logger);
    delete logger;
    logger = NULL;
  }
  if (cluster_logger) {
    if (cluster_logger_registered)
      cct->get_perfcounters_collection()->remove(cluster_logger);
    delete cluster_logger;
    cluster_logger = NULL;
  }
}

void Monitor::wait_for_paxos_write()
{
  if (paxos->is_writing() || paxos->is_writing_previous()) {
    dout(10) << __func__ << " flushing pending write" << dendl;
    lock.Unlock();
    store->flush();
    lock.Lock();
    dout(10) << __func__ << " flushed pending write" << dendl;
  }
}

void Monitor::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 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 << " is new leader, mons " << get_quorum_names()
      << " in quorum (ranks " << quorum << ")";

  set_leader_commands(get_local_commands(mon_features));

  paxos->leader_init();
  // NOTE: tell monmap monitor first.  This is important for the
  // bootstrap case to ensure that the very first paxos proposal
  // codifies the monmap.  Otherwise any manner of chaos can ensue
  // when monitors are call elections or participating in a paxos
  // round without agreeing on who the participants are.
  monmon()->election_finished();
  _finish_svc_election();
  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();

    // Freshen the health status before doing health_to_clog in case
    // our just-completed election changed the health
    healthmon()->wait_for_active_ctx(new FunctionContext([this](int r){
      dout(20) << "healthmon now active" << dendl;
      healthmon()->tick();
      if (healthmon()->is_proposing()) {
        dout(20) << __func__ << " healthmon proposing, waiting" << dendl;
        healthmon()->wait_for_finished_proposal(nullptr, new C_MonContext(this,
              [this](int r){
                assert(lock.is_locked_by_me());
                do_health_to_clog_interval();
              }));

      } else {
        do_health_to_clog_interval();
      }
    }));

    scrub_event_start();
  }
}

void Monitor::lose_election(epoch_t epoch, set<int> &q, int l,
                            uint64_t features,
                            const mon_feature_t& mon_features)
{
  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;
	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> &param_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);
  if (status != HEALTH_OK) {
    clog->health(status) << other << " " << ss.str();
  }

  dout(10) << __func__ << " from " << other << " ts " << m->timestamp
	   << " delta " << delta << " skew_bound " << skew_bound
	   << " latency " << latency << dendl;

  timecheck_skews[other] = skew_bound;

  timecheck_acks++;
  if (timecheck_acks == quorum.size()) {
    dout(10) << __func__ << " got pongs from everybody ("
             << timecheck_acks << " total)" << dendl;
    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|O_CLOEXEC, 0600);
  if (fd < 0) {
    err = -errno;
    dout(0) << __func__ << " failed to open " << os.str() 
	    << ": " << cpp_strerror(err) << dendl;
    return err;
  }

  err = bl.write_fd(fd);
  if (!err)
    ::fsync(fd);
  VOID_TEMP_FAILURE_RETRY(::close(fd));

  return err;
}

void Monitor::extract_save_mon_key(KeyRing& keyring)
{
  EntityName mon_name;
  mon_name.set_type(CEPH_ENTITY_TYPE_MON);
  EntityAuth mon_key;
  if (keyring.get_auth(mon_name, mon_key)) {
    dout(10) << "extract_save_mon_key moving mon. key to separate keyring" << dendl;
    KeyRing pkey;
    pkey.add(mon_name, mon_key);
    bufferlist bl;
    pkey.encode_plaintext(bl);
    write_default_keyring(bl);
    keyring.remove(mon_name);
  }
}

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,
				   std::unique_ptr<AuthAuthorizerChallenge> *challenge)
{
  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, challenge, 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;
}