Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1760 insertions, 0 deletions

diff --git a/src/mon/MonClient.cc b/src/mon/MonClient.cc
new file mode 100644
index 00000000..a0a6acd1
--- /dev/null
+++ b/src/mon/MonClient.cc
@@ -0,0 +1,1760 @@
+// -*- 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 <algorithm>
+#include <iterator>
+#include <random>
+
+#include "include/scope_guard.h"
+#include "include/stringify.h"
+
+#include "messages/MMonGetMap.h"
+#include "messages/MMonGetVersion.h"
+#include "messages/MMonGetVersionReply.h"
+#include "messages/MMonMap.h"
+#include "messages/MConfig.h"
+#include "messages/MGetConfig.h"
+#include "messages/MAuth.h"
+#include "messages/MLogAck.h"
+#include "messages/MAuthReply.h"
+#include "messages/MMonCommand.h"
+#include "messages/MMonCommandAck.h"
+#include "messages/MPing.h"
+
+#include "messages/MMonSubscribe.h"
+#include "messages/MMonSubscribeAck.h"
+#include "common/errno.h"
+#include "common/hostname.h"
+#include "common/LogClient.h"
+
+#include "MonClient.h"
+#include "MonMap.h"
+
+#include "auth/Auth.h"
+#include "auth/KeyRing.h"
+#include "auth/AuthClientHandler.h"
+#include "auth/AuthMethodList.h"
+#include "auth/AuthRegistry.h"
+#include "auth/RotatingKeyRing.h"
+
+#define dout_subsys ceph_subsys_monc
+#undef dout_prefix
+#define dout_prefix *_dout << "monclient" << (_hunting() ? "(hunting)":"") << ": "
+
+MonClient::MonClient(CephContext *cct_) :
+  Dispatcher(cct_),
+  AuthServer(cct_),
+  messenger(NULL),
+  monc_lock("MonClient::monc_lock"),
+  timer(cct_, monc_lock),
+  finisher(cct_),
+  initialized(false),
+  log_client(NULL),
+  more_log_pending(false),
+  want_monmap(true),
+  had_a_connection(false),
+  reopen_interval_multiplier(
+    cct_->_conf.get_val<double>("mon_client_hunt_interval_min_multiple")),
+  last_mon_command_tid(0),
+  version_req_id(0)
+{}
+
+MonClient::~MonClient()
+{
+}
+
+int MonClient::build_initial_monmap()
+{
+  ldout(cct, 10) << __func__ << dendl;
+  int r = monmap.build_initial(cct, false, cerr);
+  ldout(cct,10) << "monmap:\n";
+  monmap.print(*_dout);
+  *_dout << dendl;
+  return r;
+}
+
+int MonClient::get_monmap()
+{
+  ldout(cct, 10) << __func__ << dendl;
+  std::lock_guard l(monc_lock);
+  
+  sub.want("monmap", 0, 0);
+  if (!_opened())
+    _reopen_session();
+
+  while (want_monmap)
+    map_cond.Wait(monc_lock);
+
+  ldout(cct, 10) << __func__ << " done" << dendl;
+  return 0;
+}
+
+int MonClient::get_monmap_and_config()
+{
+  ldout(cct, 10) << __func__ << dendl;
+  ceph_assert(!messenger);
+
+  int tries = 10;
+
+  utime_t interval;
+  interval.set_from_double(cct->_conf->mon_client_hunt_interval);
+
+  cct->init_crypto();
+  auto shutdown_crypto = make_scope_guard([this] {
+    cct->shutdown_crypto();
+  });
+
+  int r = build_initial_monmap();
+  if (r < 0) {
+    lderr(cct) << __func__ << " cannot identify monitors to contact" << dendl;
+    return r;
+  }
+
+  messenger = Messenger::create_client_messenger(
+    cct, "temp_mon_client");
+  ceph_assert(messenger);
+  messenger->add_dispatcher_head(this);
+  messenger->start();
+  auto shutdown_msgr = make_scope_guard([this] {
+    messenger->shutdown();
+    messenger->wait();
+    delete messenger;
+    messenger = nullptr;
+    if (!monmap.fsid.is_zero()) {
+      cct->_conf.set_val("fsid", stringify(monmap.fsid));
+    }
+  });
+
+  while (tries-- > 0) {
+    r = init();
+    if (r < 0) {
+      return r;
+    }
+    r = authenticate(cct->_conf->client_mount_timeout);
+    if (r == -ETIMEDOUT) {
+      shutdown();
+      continue;
+    }
+    if (r < 0) {
+      break;
+    }
+    {
+      std::lock_guard l(monc_lock);
+      if (monmap.get_epoch() &&
+	  !monmap.persistent_features.contains_all(
+	    ceph::features::mon::FEATURE_MIMIC)) {
+	ldout(cct,10) << __func__ << " pre-mimic monitor, no config to fetch"
+		      << dendl;
+	r = 0;
+	break;
+      }
+      while ((!got_config || monmap.get_epoch() == 0) && r == 0) {
+	ldout(cct,20) << __func__ << " waiting for monmap|config" << dendl;
+	r = map_cond.WaitInterval(monc_lock, interval);
+      }
+      if (got_config) {
+	ldout(cct,10) << __func__ << " success" << dendl;
+	r = 0;
+	break;
+      }
+    }
+    lderr(cct) << __func__ << " failed to get config" << dendl;
+    shutdown();
+    continue;
+  }
+
+  shutdown();
+  return r;
+}
+
+
+/**
+ * Ping the monitor with id @p mon_id and set the resulting reply in
+ * the provided @p result_reply, if this last parameter is not NULL.
+ *
+ * So that we don't rely on the MonClient's default messenger, set up
+ * during connect(), we create our own messenger to comunicate with the
+ * specified monitor.  This is advantageous in the following ways:
+ *
+ * - Isolate the ping procedure from the rest of the MonClient's operations,
+ *   allowing us to not acquire or manage the big monc_lock, thus not
+ *   having to block waiting for some other operation to finish before we
+ *   can proceed.
+ *   * for instance, we can ping mon.FOO even if we are currently hunting
+ *     or blocked waiting for auth to complete with mon.BAR.
+ *
+ * - Ping a monitor prior to establishing a connection (using connect())
+ *   and properly establish the MonClient's messenger.  This frees us
+ *   from dealing with the complex foo that happens in connect().
+ *
+ * We also don't rely on MonClient as a dispatcher for this messenger,
+ * unlike what happens with the MonClient's default messenger.  This allows
+ * us to sandbox the whole ping, having it much as a separate entity in
+ * the MonClient class, considerably simplifying the handling and dispatching
+ * of messages without needing to consider monc_lock.
+ *
+ * Current drawback is that we will establish a messenger for each ping
+ * we want to issue, instead of keeping a single messenger instance that
+ * would be used for all pings.
+ */
+int MonClient::ping_monitor(const string &mon_id, string *result_reply)
+{
+  ldout(cct, 10) << __func__ << dendl;
+
+  string new_mon_id;
+  if (monmap.contains("noname-"+mon_id)) {
+    new_mon_id = "noname-"+mon_id;
+  } else {
+    new_mon_id = mon_id;
+  }
+
+  if (new_mon_id.empty()) {
+    ldout(cct, 10) << __func__ << " specified mon id is empty!" << dendl;
+    return -EINVAL;
+  } else if (!monmap.contains(new_mon_id)) {
+    ldout(cct, 10) << __func__ << " no such monitor 'mon." << new_mon_id << "'"
+                   << dendl;
+    return -ENOENT;
+  }
+
+  // N.B. monc isn't initialized
+
+  auth_registry.refresh_config();
+
+  KeyRing keyring;
+  keyring.from_ceph_context(cct);
+  RotatingKeyRing rkeyring(cct, cct->get_module_type(), &keyring);
+
+  MonClientPinger *pinger = new MonClientPinger(cct,
+						&rkeyring,
+						result_reply);
+
+  Messenger *smsgr = Messenger::create_client_messenger(cct, "temp_ping_client");
+  smsgr->add_dispatcher_head(pinger);
+  smsgr->set_auth_client(pinger);
+  smsgr->start();
+
+  ConnectionRef con = smsgr->connect_to_mon(monmap.get_addrs(new_mon_id));
+  ldout(cct, 10) << __func__ << " ping mon." << new_mon_id
+                 << " " << con->get_peer_addr() << dendl;
+
+  pinger->mc.reset(new MonConnection(cct, con, 0, &auth_registry));
+  pinger->mc->start(monmap.get_epoch(), entity_name);
+  con->send_message(new MPing);
+
+  pinger->lock.Lock();
+  int ret = pinger->wait_for_reply(cct->_conf->mon_client_ping_timeout);
+  if (ret == 0) {
+    ldout(cct,10) << __func__ << " got ping reply" << dendl;
+  } else {
+    ret = -ret;
+  }
+  pinger->lock.Unlock();
+
+  con->mark_down();
+  pinger->mc.reset();
+  smsgr->shutdown();
+  smsgr->wait();
+  delete smsgr;
+  delete pinger;
+  return ret;
+}
+
+bool MonClient::ms_dispatch(Message *m)
+{
+  // we only care about these message types
+  switch (m->get_type()) {
+  case CEPH_MSG_MON_MAP:
+  case CEPH_MSG_AUTH_REPLY:
+  case CEPH_MSG_MON_SUBSCRIBE_ACK:
+  case CEPH_MSG_MON_GET_VERSION_REPLY:
+  case MSG_MON_COMMAND_ACK:
+  case MSG_LOGACK:
+  case MSG_CONFIG:
+    break;
+  case CEPH_MSG_PING:
+    m->put();
+    return true;
+  default:
+    return false;
+  }
+
+  std::lock_guard lock(monc_lock);
+
+  if (_hunting()) {
+    auto p = _find_pending_con(m->get_connection());
+    if (p == pending_cons.end()) {
+      // ignore any messages outside hunting sessions
+      ldout(cct, 10) << "discarding stray monitor message " << *m << dendl;
+      m->put();
+      return true;
+    }
+  } else if (!active_con || active_con->get_con() != m->get_connection()) {
+    // ignore any messages outside our session(s)
+    ldout(cct, 10) << "discarding stray monitor message " << *m << dendl;
+    m->put();
+    return true;
+  }
+
+  switch (m->get_type()) {
+  case CEPH_MSG_MON_MAP:
+    handle_monmap(static_cast<MMonMap*>(m));
+    if (passthrough_monmap) {
+      return false;
+    } else {
+      m->put();
+    }
+    break;
+  case CEPH_MSG_AUTH_REPLY:
+    handle_auth(static_cast<MAuthReply*>(m));
+    break;
+  case CEPH_MSG_MON_SUBSCRIBE_ACK:
+    handle_subscribe_ack(static_cast<MMonSubscribeAck*>(m));
+    break;
+  case CEPH_MSG_MON_GET_VERSION_REPLY:
+    handle_get_version_reply(static_cast<MMonGetVersionReply*>(m));
+    break;
+  case MSG_MON_COMMAND_ACK:
+    handle_mon_command_ack(static_cast<MMonCommandAck*>(m));
+    break;
+  case MSG_LOGACK:
+    if (log_client) {
+      log_client->handle_log_ack(static_cast<MLogAck*>(m));
+      m->put();
+      if (more_log_pending) {
+	send_log();
+      }
+    } else {
+      m->put();
+    }
+    break;
+  case MSG_CONFIG:
+    handle_config(static_cast<MConfig*>(m));
+    break;
+  }
+  return true;
+}
+
+void MonClient::send_log(bool flush)
+{
+  if (log_client) {
+    Message *lm = log_client->get_mon_log_message(flush);
+    if (lm)
+      _send_mon_message(lm);
+    more_log_pending = log_client->are_pending();
+  }
+}
+
+void MonClient::flush_log()
+{
+  std::lock_guard l(monc_lock);
+  send_log();
+}
+
+/* Unlike all the other message-handling functions, we don't put away a reference
+* because we want to support MMonMap passthrough to other Dispatchers. */
+void MonClient::handle_monmap(MMonMap *m)
+{
+  ldout(cct, 10) << __func__ << " " << *m << dendl;
+  auto con_addrs = m->get_source_addrs();
+  string old_name = monmap.get_name(con_addrs);
+
+  // NOTE: we're not paying attention to the epoch, here.
+  auto p = m->monmapbl.cbegin();
+  decode(monmap, p);
+
+  ldout(cct, 10) << " got monmap " << monmap.epoch
+		 << " from mon." << old_name
+		 << " (according to old e" << monmap.get_epoch() << ")"
+ 		 << dendl;
+  ldout(cct, 10) << "dump:\n";
+  monmap.print(*_dout);
+  *_dout << dendl;
+
+  if (old_name.size() == 0) {
+    ldout(cct,10) << " can't identify which mon we were connected to" << dendl;
+    _reopen_session();
+  } else {
+    int new_rank = monmap.get_rank(m->get_source_addr());
+    if (new_rank < 0) {
+      ldout(cct, 10) << "mon." << new_rank << " at " << m->get_source_addrs()
+		     << " went away" << dendl;
+      // can't find the mon we were talking to (above)
+      _reopen_session();
+    } else if (messenger->should_use_msgr2() &&
+	       monmap.get_addrs(new_rank).has_msgr2() &&
+	       !con_addrs.has_msgr2()) {
+      ldout(cct,1) << " mon." << new_rank << " has (v2) addrs "
+		   << monmap.get_addrs(new_rank) << " but i'm connected to "
+		   << con_addrs << ", reconnecting" << dendl;
+      _reopen_session();
+    }
+  }
+
+  cct->set_mon_addrs(monmap);
+
+  sub.got("monmap", monmap.get_epoch());
+  map_cond.Signal();
+  want_monmap = false;
+
+  if (authenticate_err == 1) {
+    _finish_auth(0);
+  }
+}
+
+void MonClient::handle_config(MConfig *m)
+{
+  ldout(cct,10) << __func__ << " " << *m << dendl;
+  finisher.queue(new FunctionContext([this, m](int r) {
+	cct->_conf.set_mon_vals(cct, m->config, config_cb);
+	if (config_notify_cb) {
+	  config_notify_cb();
+	}
+	m->put();
+      }));
+  got_config = true;
+  map_cond.Signal();
+}
+
+// ----------------------
+
+int MonClient::init()
+{
+  ldout(cct, 10) << __func__ << dendl;
+
+  entity_name = cct->_conf->name;
+
+  auth_registry.refresh_config();
+
+  std::lock_guard l(monc_lock);
+  keyring.reset(new KeyRing);
+  if (auth_registry.is_supported_method(messenger->get_mytype(),
+					CEPH_AUTH_CEPHX)) {
+    // this should succeed, because auth_registry just checked!
+    int r = keyring->from_ceph_context(cct);
+    if (r != 0) {
+      // but be somewhat graceful in case there was a race condition
+      lderr(cct) << "keyring not found" << dendl;
+      return r;
+    }
+  }
+  if (!auth_registry.any_supported_methods(messenger->get_mytype())) {
+    return -ENOENT;
+  }
+
+  rotating_secrets.reset(
+    new RotatingKeyRing(cct, cct->get_module_type(), keyring.get()));
+
+  initialized = true;
+
+  messenger->set_auth_client(this);
+  messenger->add_dispatcher_head(this);
+
+  timer.init();
+  finisher.start();
+  schedule_tick();
+
+  return 0;
+}
+
+void MonClient::shutdown()
+{
+  ldout(cct, 10) << __func__ << dendl;
+  monc_lock.Lock();
+  stopping = true;
+  while (!version_requests.empty()) {
+    version_requests.begin()->second->context->complete(-ECANCELED);
+    ldout(cct, 20) << __func__ << " canceling and discarding version request "
+		   << version_requests.begin()->second << dendl;
+    delete version_requests.begin()->second;
+    version_requests.erase(version_requests.begin());
+  }
+  while (!mon_commands.empty()) {
+    auto tid = mon_commands.begin()->first;
+    _cancel_mon_command(tid);
+  }
+  while (!waiting_for_session.empty()) {
+    ldout(cct, 20) << __func__ << " discarding pending message " << *waiting_for_session.front() << dendl;
+    waiting_for_session.front()->put();
+    waiting_for_session.pop_front();
+  }
+
+  active_con.reset();
+  pending_cons.clear();
+
+  auth.reset();
+  global_id = 0;
+  authenticate_err = 0;
+  authenticated = false;
+
+  monc_lock.Unlock();
+
+  if (initialized) {
+    finisher.wait_for_empty();
+    finisher.stop();
+    initialized = false;
+  }
+  monc_lock.Lock();
+  timer.shutdown();
+  stopping = false;
+  monc_lock.Unlock();
+}
+
+int MonClient::authenticate(double timeout)
+{
+  std::lock_guard lock(monc_lock);
+
+  if (active_con) {
+    ldout(cct, 5) << "already authenticated" << dendl;
+    return 0;
+  }
+  sub.want("monmap", monmap.get_epoch() ? monmap.get_epoch() + 1 : 0, 0);
+  sub.want("config", 0, 0);
+  if (!_opened())
+    _reopen_session();
+
+  utime_t until = ceph_clock_now();
+  until += timeout;
+  if (timeout > 0.0)
+    ldout(cct, 10) << "authenticate will time out at " << until << dendl;
+  authenticate_err = 1;  // == in progress
+  while (!active_con && authenticate_err >= 0) {
+    if (timeout > 0.0) {
+      int r = auth_cond.WaitUntil(monc_lock, until);
+      if (r == ETIMEDOUT && !active_con) {
+	ldout(cct, 0) << "authenticate timed out after " << timeout << dendl;
+	authenticate_err = -r;
+      }
+    } else {
+      auth_cond.Wait(monc_lock);
+    }
+  }
+
+  if (active_con) {
+    ldout(cct, 5) << __func__ << " success, global_id "
+		  << active_con->get_global_id() << dendl;
+    // active_con should not have been set if there was an error
+    ceph_assert(authenticate_err >= 0);
+    authenticated = true;
+  }
+
+  if (authenticate_err < 0 && auth_registry.no_keyring_disabled_cephx()) {
+    lderr(cct) << __func__ << " NOTE: no keyring found; disabled cephx authentication" << dendl;
+  }
+
+  return authenticate_err;
+}
+
+void MonClient::handle_auth(MAuthReply *m)
+{
+  ceph_assert(monc_lock.is_locked());
+  if (!_hunting()) {
+    std::swap(active_con->get_auth(), auth);
+    int ret = active_con->authenticate(m);
+    m->put();
+    std::swap(auth, active_con->get_auth());
+    if (global_id != active_con->get_global_id()) {
+      lderr(cct) << __func__ << " peer assigned me a different global_id: "
+		 << active_con->get_global_id() << dendl;
+    }
+    if (ret != -EAGAIN) {
+      _finish_auth(ret);
+    }
+    return;
+  }
+
+  // hunting
+  auto found = _find_pending_con(m->get_connection());
+  ceph_assert(found != pending_cons.end());
+  int auth_err = found->second.handle_auth(m, entity_name, want_keys,
+					   rotating_secrets.get());
+  m->put();
+  if (auth_err == -EAGAIN) {
+    return;
+  }
+  if (auth_err) {
+    pending_cons.erase(found);
+    if (!pending_cons.empty()) {
+      // keep trying with pending connections
+      return;
+    }
+    // the last try just failed, give up.
+  } else {
+    auto& mc = found->second;
+    ceph_assert(mc.have_session());
+    active_con.reset(new MonConnection(std::move(mc)));
+    pending_cons.clear();
+  }
+
+  _finish_hunting(auth_err);
+  _finish_auth(auth_err);
+}
+
+void MonClient::_finish_auth(int auth_err)
+{
+  ldout(cct,10) << __func__ << " " << auth_err << dendl;
+  authenticate_err = auth_err;
+  // _resend_mon_commands() could _reopen_session() if the connected mon is not
+  // the one the MonCommand is targeting.
+  if (!auth_err && active_con) {
+    ceph_assert(auth);
+    _check_auth_tickets();
+  }
+  auth_cond.SignalAll();
+
+  if (!auth_err) {
+    Context *cb = nullptr;
+    if (session_established_context) {
+      cb = session_established_context.release();
+    }
+    if (cb) {
+      monc_lock.Unlock();
+      cb->complete(0);
+      monc_lock.Lock();
+    }
+  }
+}
+
+// ---------
+
+void MonClient::_send_mon_message(Message *m)
+{
+  ceph_assert(monc_lock.is_locked());
+  if (active_con) {
+    auto cur_con = active_con->get_con();
+    ldout(cct, 10) << "_send_mon_message to mon."
+		   << monmap.get_name(cur_con->get_peer_addr())
+		   << " at " << cur_con->get_peer_addr() << dendl;
+    cur_con->send_message(m);
+  } else {
+    waiting_for_session.push_back(m);
+  }
+}
+
+void MonClient::_reopen_session(int rank)
+{
+  ceph_assert(monc_lock.is_locked());
+  ldout(cct, 10) << __func__ << " rank " << rank << dendl;
+
+  active_con.reset();
+  pending_cons.clear();
+
+  _start_hunting();
+
+  if (rank >= 0) {
+    _add_conn(rank);
+  } else {
+    _add_conns();
+  }
+
+  // throw out old queued messages
+  while (!waiting_for_session.empty()) {
+    waiting_for_session.front()->put();
+    waiting_for_session.pop_front();
+  }
+
+  // throw out version check requests
+  while (!version_requests.empty()) {
+    finisher.queue(version_requests.begin()->second->context, -EAGAIN);
+    delete version_requests.begin()->second;
+    version_requests.erase(version_requests.begin());
+  }
+
+  for (auto& c : pending_cons) {
+    c.second.start(monmap.get_epoch(), entity_name);
+  }
+
+  if (sub.reload()) {
+    _renew_subs();
+  }
+}
+
+MonConnection& MonClient::_add_conn(unsigned rank)
+{
+  auto peer = monmap.get_addrs(rank);
+  auto conn = messenger->connect_to_mon(peer);
+  MonConnection mc(cct, conn, global_id, &auth_registry);
+  if (auth) {
+    mc.get_auth().reset(auth->clone());
+  }
+  auto inserted = pending_cons.insert(make_pair(peer, move(mc)));
+  ldout(cct, 10) << "picked mon." << monmap.get_name(rank)
+                 << " con " << conn
+                 << " addr " << peer
+                 << dendl;
+  return inserted.first->second;
+}
+
+void MonClient::_add_conns()
+{
+  uint16_t min_priority = std::numeric_limits<uint16_t>::max();
+  for (const auto& m : monmap.mon_info) {
+    if (m.second.priority < min_priority) {
+      min_priority = m.second.priority;
+    }
+  }
+  vector<unsigned> ranks;
+  for (const auto& m : monmap.mon_info) {
+    if (m.second.priority == min_priority) {
+      ranks.push_back(monmap.get_rank(m.first));
+    }
+  }
+  std::random_device rd;
+  std::mt19937 rng(rd());
+  std::shuffle(ranks.begin(), ranks.end(), rng);
+  unsigned n = cct->_conf->mon_client_hunt_parallel;
+  if (n == 0 || n > ranks.size()) {
+    n = ranks.size();
+  }
+  for (unsigned i = 0; i < n; i++) {
+    _add_conn(ranks[i]);
+  }
+}
+
+bool MonClient::ms_handle_reset(Connection *con)
+{
+  std::lock_guard lock(monc_lock);
+
+  if (con->get_peer_type() != CEPH_ENTITY_TYPE_MON)
+    return false;
+
+  if (_hunting()) {
+    if (pending_cons.count(con->get_peer_addrs())) {
+      ldout(cct, 10) << __func__ << " hunted mon " << con->get_peer_addrs()
+		     << dendl;
+    } else {
+      ldout(cct, 10) << __func__ << " stray mon " << con->get_peer_addrs()
+		     << dendl;
+    }
+    return true;
+  } else {
+    if (active_con && con == active_con->get_con()) {
+      ldout(cct, 10) << __func__ << " current mon " << con->get_peer_addrs()
+		     << dendl;
+      _reopen_session();
+      return false;
+    } else {
+      ldout(cct, 10) << "ms_handle_reset stray mon " << con->get_peer_addrs()
+		     << dendl;
+      return true;
+    }
+  }
+}
+
+bool MonClient::_opened() const
+{
+  ceph_assert(monc_lock.is_locked());
+  return active_con || _hunting();
+}
+
+bool MonClient::_hunting() const
+{
+  return !pending_cons.empty();
+}
+
+void MonClient::_start_hunting()
+{
+  ceph_assert(!_hunting());
+  // adjust timeouts if necessary
+  if (!had_a_connection)
+    return;
+  reopen_interval_multiplier *= cct->_conf->mon_client_hunt_interval_backoff;
+  if (reopen_interval_multiplier >
+      cct->_conf->mon_client_hunt_interval_max_multiple) {
+    reopen_interval_multiplier =
+      cct->_conf->mon_client_hunt_interval_max_multiple;
+  }
+}
+
+void MonClient::_finish_hunting(int auth_err)
+{
+  ldout(cct,10) << __func__ << " " << auth_err << dendl;
+  ceph_assert(monc_lock.is_locked());
+  // the pending conns have been cleaned.
+  ceph_assert(!_hunting());
+  if (active_con) {
+    auto con = active_con->get_con();
+    ldout(cct, 1) << "found mon."
+		  << monmap.get_name(con->get_peer_addr())
+		  << dendl;
+  } else {
+    ldout(cct, 1) << "no mon sessions established" << dendl;
+  }
+
+  had_a_connection = true;
+  _un_backoff();
+
+  if (!auth_err) {
+    last_rotating_renew_sent = utime_t();
+    while (!waiting_for_session.empty()) {
+      _send_mon_message(waiting_for_session.front());
+      waiting_for_session.pop_front();
+    }
+    _resend_mon_commands();
+    send_log(true);
+    if (active_con) {
+      auth = std::move(active_con->get_auth());
+      if (global_id && global_id != active_con->get_global_id()) {
+	lderr(cct) << __func__ << " global_id changed from " << global_id
+		   << " to " << active_con->get_global_id() << dendl;
+      }
+      global_id = active_con->get_global_id();
+    }
+  }
+}
+
+void MonClient::tick()
+{
+  ldout(cct, 10) << __func__ << dendl;
+
+  auto reschedule_tick = make_scope_guard([this] {
+      schedule_tick();
+    });
+
+  _check_auth_tickets();
+  
+  if (_hunting()) {
+    ldout(cct, 1) << "continuing hunt" << dendl;
+    return _reopen_session();
+  } else if (active_con) {
+    // just renew as needed
+    utime_t now = ceph_clock_now();
+    auto cur_con = active_con->get_con();
+    if (!cur_con->has_feature(CEPH_FEATURE_MON_STATEFUL_SUB)) {
+      const bool maybe_renew = sub.need_renew();
+      ldout(cct, 10) << "renew subs? -- " << (maybe_renew ? "yes" : "no")
+		     << dendl;
+      if (maybe_renew) {
+	_renew_subs();
+      }
+    }
+
+    cur_con->send_keepalive();
+
+    if (cct->_conf->mon_client_ping_timeout > 0 &&
+	cur_con->has_feature(CEPH_FEATURE_MSGR_KEEPALIVE2)) {
+      utime_t lk = cur_con->get_last_keepalive_ack();
+      utime_t interval = now - lk;
+      if (interval > cct->_conf->mon_client_ping_timeout) {
+	ldout(cct, 1) << "no keepalive since " << lk << " (" << interval
+		      << " seconds), reconnecting" << dendl;
+	return _reopen_session();
+      }
+      send_log();
+    }
+
+    _un_backoff();
+  }
+}
+
+void MonClient::_un_backoff()
+{
+  // un-backoff our reconnect interval
+  reopen_interval_multiplier = std::max(
+    cct->_conf.get_val<double>("mon_client_hunt_interval_min_multiple"),
+    reopen_interval_multiplier /
+    cct->_conf.get_val<double>("mon_client_hunt_interval_backoff"));
+  ldout(cct, 20) << __func__ << " reopen_interval_multipler now "
+		 << reopen_interval_multiplier << dendl;
+}
+
+void MonClient::schedule_tick()
+{
+  auto do_tick = make_lambda_context([this]() { tick(); });
+  if (!is_connected()) {
+    // start another round of hunting
+    const auto hunt_interval = (cct->_conf->mon_client_hunt_interval *
+				reopen_interval_multiplier);
+    timer.add_event_after(hunt_interval, do_tick);
+  } else {
+    // keep in touch
+    timer.add_event_after(cct->_conf->mon_client_ping_interval, do_tick);
+  }
+}
+
+// ---------
+
+void MonClient::_renew_subs()
+{
+  ceph_assert(monc_lock.is_locked());
+  if (!sub.have_new()) {
+    ldout(cct, 10) << __func__ << " - empty" << dendl;
+    return;
+  }
+
+  ldout(cct, 10) << __func__ << dendl;
+  if (!_opened())
+    _reopen_session();
+  else {
+    MMonSubscribe *m = new MMonSubscribe;
+    m->what = sub.get_subs();
+    m->hostname = ceph_get_short_hostname();
+    _send_mon_message(m);
+    sub.renewed();
+  }
+}
+
+void MonClient::handle_subscribe_ack(MMonSubscribeAck *m)
+{
+  sub.acked(m->interval);
+  m->put();
+}
+
+int MonClient::_check_auth_tickets()
+{
+  ceph_assert(monc_lock.is_locked());
+  if (active_con && auth) {
+    if (auth->need_tickets()) {
+      ldout(cct, 10) << __func__ << " getting new tickets!" << dendl;
+      MAuth *m = new MAuth;
+      m->protocol = auth->get_protocol();
+      auth->prepare_build_request();
+      auth->build_request(m->auth_payload);
+      _send_mon_message(m);
+    }
+
+    _check_auth_rotating();
+  }
+  return 0;
+}
+
+int MonClient::_check_auth_rotating()
+{
+  ceph_assert(monc_lock.is_locked());
+  if (!rotating_secrets ||
+      !auth_principal_needs_rotating_keys(entity_name)) {
+    ldout(cct, 20) << "_check_auth_rotating not needed by " << entity_name << dendl;
+    return 0;
+  }
+
+  if (!active_con || !auth) {
+    ldout(cct, 10) << "_check_auth_rotating waiting for auth session" << dendl;
+    return 0;
+  }
+
+  utime_t now = ceph_clock_now();
+  utime_t cutoff = now;
+  cutoff -= std::min(30.0, cct->_conf->auth_service_ticket_ttl / 4.0);
+  utime_t issued_at_lower_bound = now;
+  issued_at_lower_bound -= cct->_conf->auth_service_ticket_ttl;
+  if (!rotating_secrets->need_new_secrets(cutoff)) {
+    ldout(cct, 10) << "_check_auth_rotating have uptodate secrets (they expire after " << cutoff << ")" << dendl;
+    rotating_secrets->dump_rotating();
+    return 0;
+  }
+
+  ldout(cct, 10) << "_check_auth_rotating renewing rotating keys (they expired before " << cutoff << ")" << dendl;
+  if (!rotating_secrets->need_new_secrets() &&
+      rotating_secrets->need_new_secrets(issued_at_lower_bound)) {
+    // the key has expired before it has been issued?
+    lderr(cct) << __func__ << " possible clock skew, rotating keys expired way too early"
+               << " (before " << issued_at_lower_bound << ")" << dendl;
+  }
+  if ((now > last_rotating_renew_sent) &&
+      double(now - last_rotating_renew_sent) < 1) {
+    ldout(cct, 10) << __func__ << " called too often (last: "
+                   << last_rotating_renew_sent << "), skipping refresh" << dendl;
+    return 0;
+  }
+  MAuth *m = new MAuth;
+  m->protocol = auth->get_protocol();
+  if (auth->build_rotating_request(m->auth_payload)) {
+    last_rotating_renew_sent = now;
+    _send_mon_message(m);
+  } else {
+    m->put();
+  }
+  return 0;
+}
+
+int MonClient::wait_auth_rotating(double timeout)
+{
+  std::lock_guard l(monc_lock);
+  utime_t now = ceph_clock_now();
+  utime_t until = now;
+  until += timeout;
+
+  // Must be initialized
+  ceph_assert(auth != nullptr);
+
+  if (auth->get_protocol() == CEPH_AUTH_NONE)
+    return 0;
+  
+  if (!rotating_secrets)
+    return 0;
+
+  while (auth_principal_needs_rotating_keys(entity_name) &&
+	 rotating_secrets->need_new_secrets(now)) {
+    if (now >= until) {
+      ldout(cct, 0) << __func__ << " timed out after " << timeout << dendl;
+      return -ETIMEDOUT;
+    }
+    ldout(cct, 10) << __func__ << " waiting (until " << until << ")" << dendl;
+    auth_cond.WaitUntil(monc_lock, until);
+    now = ceph_clock_now();
+  }
+  ldout(cct, 10) << __func__ << " done" << dendl;
+  return 0;
+}
+
+// ---------
+
+void MonClient::_send_command(MonCommand *r)
+{
+  ++r->send_attempts;
+
+  entity_addr_t peer;
+  if (active_con) {
+    peer = active_con->get_con()->get_peer_addr();
+  }
+
+  if (r->target_rank >= 0 &&
+      r->target_rank != monmap.get_rank(peer)) {
+    if (r->send_attempts > cct->_conf->mon_client_directed_command_retry) {
+      _finish_command(r, -ENXIO, "mon unavailable");
+      return;
+    }
+    ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd
+		   << " wants rank " << r->target_rank
+		   << ", reopening session"
+		   << dendl;
+    if (r->target_rank >= (int)monmap.size()) {
+      ldout(cct, 10) << " target " << r->target_rank << " >= max mon " << monmap.size() << dendl;
+      _finish_command(r, -ENOENT, "mon rank dne");
+      return;
+    }
+    _reopen_session(r->target_rank);
+    return;
+  }
+
+  if (r->target_name.length() &&
+      r->target_name != monmap.get_name(peer)) {
+    if (r->send_attempts > cct->_conf->mon_client_directed_command_retry) {
+      _finish_command(r, -ENXIO, "mon unavailable");
+      return;
+    }
+    ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd
+		   << " wants mon " << r->target_name
+		   << ", reopening session"
+		   << dendl;
+    if (!monmap.contains(r->target_name)) {
+      ldout(cct, 10) << " target " << r->target_name << " not present in monmap" << dendl;
+      _finish_command(r, -ENOENT, "mon dne");
+      return;
+    }
+    _reopen_session(monmap.get_rank(r->target_name));
+    return;
+  }
+
+  ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
+  MMonCommand *m = new MMonCommand(monmap.fsid);
+  m->set_tid(r->tid);
+  m->cmd = r->cmd;
+  m->set_data(r->inbl);
+  _send_mon_message(m);
+  return;
+}
+
+void MonClient::_resend_mon_commands()
+{
+  // resend any requests
+  map<uint64_t,MonCommand*>::iterator p = mon_commands.begin();
+  while (p != mon_commands.end()) {
+    auto cmd = p->second;
+    ++p;
+    _send_command(cmd); // might remove cmd from mon_commands
+  }
+}
+
+void MonClient::handle_mon_command_ack(MMonCommandAck *ack)
+{
+  MonCommand *r = NULL;
+  uint64_t tid = ack->get_tid();
+
+  if (tid == 0 && !mon_commands.empty()) {
+    r = mon_commands.begin()->second;
+    ldout(cct, 10) << __func__ << " has tid 0, assuming it is " << r->tid << dendl;
+  } else {
+    map<uint64_t,MonCommand*>::iterator p = mon_commands.find(tid);
+    if (p == mon_commands.end()) {
+      ldout(cct, 10) << __func__ << " " << ack->get_tid() << " not found" << dendl;
+      ack->put();
+      return;
+    }
+    r = p->second;
+  }
+
+  ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
+  if (r->poutbl)
+    r->poutbl->claim(ack->get_data());
+  _finish_command(r, ack->r, ack->rs);
+  ack->put();
+}
+
+int MonClient::_cancel_mon_command(uint64_t tid)
+{
+  ceph_assert(monc_lock.is_locked());
+
+  map<ceph_tid_t, MonCommand*>::iterator it = mon_commands.find(tid);
+  if (it == mon_commands.end()) {
+    ldout(cct, 10) << __func__ << " tid " << tid << " dne" << dendl;
+    return -ENOENT;
+  }
+
+  ldout(cct, 10) << __func__ << " tid " << tid << dendl;
+
+  MonCommand *cmd = it->second;
+  _finish_command(cmd, -ETIMEDOUT, "");
+  return 0;
+}
+
+void MonClient::_finish_command(MonCommand *r, int ret, string rs)
+{
+  ldout(cct, 10) << __func__ << " " << r->tid << " = " << ret << " " << rs << dendl;
+  if (r->prval)
+    *(r->prval) = ret;
+  if (r->prs)
+    *(r->prs) = rs;
+  if (r->onfinish)
+    finisher.queue(r->onfinish, ret);
+  mon_commands.erase(r->tid);
+  delete r;
+}
+
+void MonClient::start_mon_command(const vector<string>& cmd,
+				 const bufferlist& inbl,
+				 bufferlist *outbl, string *outs,
+				 Context *onfinish)
+{
+  ldout(cct,10) << __func__ << " cmd=" << cmd << dendl;
+  std::lock_guard l(monc_lock);
+  if (!initialized || stopping) {
+    if (onfinish) {
+      onfinish->complete(-ECANCELED);
+    }
+    return;
+  }
+  MonCommand *r = new MonCommand(++last_mon_command_tid);
+  r->cmd = cmd;
+  r->inbl = inbl;
+  r->poutbl = outbl;
+  r->prs = outs;
+  r->onfinish = onfinish;
+  if (cct->_conf->rados_mon_op_timeout > 0) {
+    class C_CancelMonCommand : public Context
+    {
+      uint64_t tid;
+      MonClient *monc;
+      public:
+      C_CancelMonCommand(uint64_t tid, MonClient *monc) : tid(tid), monc(monc) {}
+      void finish(int r) override {
+	monc->_cancel_mon_command(tid);
+      }
+    };
+    r->ontimeout = new C_CancelMonCommand(r->tid, this);
+    timer.add_event_after(cct->_conf->rados_mon_op_timeout, r->ontimeout);
+  }
+  mon_commands[r->tid] = r;
+  _send_command(r);
+}
+
+void MonClient::start_mon_command(const string &mon_name,
+				 const vector<string>& cmd,
+				 const bufferlist& inbl,
+				 bufferlist *outbl, string *outs,
+				 Context *onfinish)
+{
+  ldout(cct,10) << __func__ << " mon." << mon_name << " cmd=" << cmd << dendl;
+  std::lock_guard l(monc_lock);
+  if (!initialized || stopping) {
+    if (onfinish) {
+      onfinish->complete(-ECANCELED);
+    }
+    return;
+  }
+  MonCommand *r = new MonCommand(++last_mon_command_tid);
+  r->target_name = mon_name;
+  r->cmd = cmd;
+  r->inbl = inbl;
+  r->poutbl = outbl;
+  r->prs = outs;
+  r->onfinish = onfinish;
+  mon_commands[r->tid] = r;
+  _send_command(r);
+}
+
+void MonClient::start_mon_command(int rank,
+				 const vector<string>& cmd,
+				 const bufferlist& inbl,
+				 bufferlist *outbl, string *outs,
+				 Context *onfinish)
+{
+  ldout(cct,10) << __func__ << " rank " << rank << " cmd=" << cmd << dendl;
+  std::lock_guard l(monc_lock);
+  if (!initialized || stopping) {
+    if (onfinish) {
+      onfinish->complete(-ECANCELED);
+    }
+    return;
+  }
+  MonCommand *r = new MonCommand(++last_mon_command_tid);
+  r->target_rank = rank;
+  r->cmd = cmd;
+  r->inbl = inbl;
+  r->poutbl = outbl;
+  r->prs = outs;
+  r->onfinish = onfinish;
+  mon_commands[r->tid] = r;
+  _send_command(r);
+}
+
+// ---------
+
+void MonClient::get_version(string map, version_t *newest, version_t *oldest, Context *onfinish)
+{
+  version_req_d *req = new version_req_d(onfinish, newest, oldest);
+  ldout(cct, 10) << "get_version " << map << " req " << req << dendl;
+  std::lock_guard l(monc_lock);
+  MMonGetVersion *m = new MMonGetVersion();
+  m->what = map;
+  m->handle = ++version_req_id;
+  version_requests[m->handle] = req;
+  _send_mon_message(m);
+}
+
+void MonClient::handle_get_version_reply(MMonGetVersionReply* m)
+{
+  ceph_assert(monc_lock.is_locked());
+  map<ceph_tid_t, version_req_d*>::iterator iter = version_requests.find(m->handle);
+  if (iter == version_requests.end()) {
+    ldout(cct, 0) << __func__ << " version request with handle " << m->handle
+		  << " not found" << dendl;
+  } else {
+    version_req_d *req = iter->second;
+    ldout(cct, 10) << __func__ << " finishing " << req << " version " << m->version << dendl;
+    version_requests.erase(iter);
+    if (req->newest)
+      *req->newest = m->version;
+    if (req->oldest)
+      *req->oldest = m->oldest_version;
+    finisher.queue(req->context, 0);
+    delete req;
+  }
+  m->put();
+}
+
+int MonClient::get_auth_request(
+  Connection *con,
+  AuthConnectionMeta *auth_meta,
+  uint32_t *auth_method,
+  std::vector<uint32_t> *preferred_modes,
+  bufferlist *bl)
+{
+  std::lock_guard l(monc_lock);
+  ldout(cct,10) << __func__ << " con " << con << " auth_method " << *auth_method
+		<< dendl;
+
+  // connection to mon?
+  if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) {
+    ceph_assert(!auth_meta->authorizer);
+    for (auto& i : pending_cons) {
+      if (i.second.is_con(con)) {
+	return i.second.get_auth_request(
+	  auth_method, preferred_modes, bl,
+	  entity_name, want_keys, rotating_secrets.get());
+      }
+    }
+    return -ENOENT;
+  }
+
+  // generate authorizer
+  if (!auth) {
+    lderr(cct) << __func__ << " but no auth handler is set up" << dendl;
+    return -EACCES;
+  }
+  auth_meta->authorizer.reset(auth->build_authorizer(con->get_peer_type()));
+  if (!auth_meta->authorizer) {
+    lderr(cct) << __func__ << " failed to build_authorizer for type "
+	       << ceph_entity_type_name(con->get_peer_type()) << dendl;
+    return -EACCES;
+  }
+  auth_meta->auth_method = auth_meta->authorizer->protocol;
+  auth_registry.get_supported_modes(con->get_peer_type(),
+				    auth_meta->auth_method,
+				    preferred_modes);
+  *bl = auth_meta->authorizer->bl;
+  return 0;
+}
+
+int MonClient::handle_auth_reply_more(
+  Connection *con,
+  AuthConnectionMeta *auth_meta,
+  const bufferlist& bl,
+  bufferlist *reply)
+{
+  std::lock_guard l(monc_lock);
+
+  if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) {
+    for (auto& i : pending_cons) {
+      if (i.second.is_con(con)) {
+	return i.second.handle_auth_reply_more(auth_meta, bl, reply);
+      }
+    }
+    return -ENOENT;
+  }
+
+  // authorizer challenges
+  if (!auth || !auth_meta->authorizer) {
+    lderr(cct) << __func__ << " no authorizer?" << dendl;
+    return -1;
+  }
+  auth_meta->authorizer->add_challenge(cct, bl);
+  *reply = auth_meta->authorizer->bl;
+  return 0;
+}
+
+int MonClient::handle_auth_done(
+  Connection *con,
+  AuthConnectionMeta *auth_meta,
+  uint64_t global_id,
+  uint32_t con_mode,
+  const bufferlist& bl,
+  CryptoKey *session_key,
+  std::string *connection_secret)
+{
+  if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) {
+    std::lock_guard l(monc_lock);
+    for (auto& i : pending_cons) {
+      if (i.second.is_con(con)) {
+	int r = i.second.handle_auth_done(
+	  auth_meta, global_id, bl,
+	  session_key, connection_secret);
+	if (r) {
+	  pending_cons.erase(i.first);
+	  if (!pending_cons.empty()) {
+	    return r;
+	  }
+	} else {
+	  active_con.reset(new MonConnection(std::move(i.second)));
+	  pending_cons.clear();
+	  ceph_assert(active_con->have_session());
+	}
+
+	_finish_hunting(r);
+	if (r || monmap.get_epoch() > 0) {
+	  _finish_auth(r);
+	}
+	return r;
+      }
+    }
+    return -ENOENT;
+  } else {
+    // verify authorizer reply
+    auto p = bl.begin();
+    if (!auth_meta->authorizer->verify_reply(p, &auth_meta->connection_secret)) {
+      ldout(cct, 0) << __func__ << " failed verifying authorizer reply"
+		    << dendl;
+      return -EACCES;
+    }
+    auth_meta->session_key = auth_meta->authorizer->session_key;
+    return 0;
+  }
+}
+
+int MonClient::handle_auth_bad_method(
+  Connection *con,
+  AuthConnectionMeta *auth_meta,
+  uint32_t old_auth_method,
+  int result,
+  const std::vector<uint32_t>& allowed_methods,
+  const std::vector<uint32_t>& allowed_modes)
+{
+  auth_meta->allowed_methods = allowed_methods;
+
+  std::lock_guard l(monc_lock);
+  if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) {
+    for (auto& i : pending_cons) {
+      if (i.second.is_con(con)) {
+	int r = i.second.handle_auth_bad_method(old_auth_method,
+						result,
+						allowed_methods,
+						allowed_modes);
+	if (r == 0) {
+	  return r; // try another method on this con
+	}
+	pending_cons.erase(i.first);
+	if (!pending_cons.empty()) {
+	  return r;  // fail this con, maybe another con will succeed
+	}
+	// fail hunt
+	_finish_hunting(r);
+	_finish_auth(r);
+	return r;
+      }
+    }
+    return -ENOENT;
+  } else {
+    // huh...
+    ldout(cct,10) << __func__ << " hmm, they didn't like " << old_auth_method
+		  << " result " << cpp_strerror(result)
+		  << " and auth is " << (auth ? auth->get_protocol() : 0)
+		  << dendl;
+    return -EACCES;
+  }
+}
+
+int MonClient::handle_auth_request(
+  Connection *con,
+  AuthConnectionMeta *auth_meta,
+  bool more,
+  uint32_t auth_method,
+  const bufferlist& payload,
+  bufferlist *reply)
+{
+  auth_meta->auth_mode = payload[0];
+  if (auth_meta->auth_mode < AUTH_MODE_AUTHORIZER ||
+      auth_meta->auth_mode > AUTH_MODE_AUTHORIZER_MAX) {
+    return -EACCES;
+  }
+  AuthAuthorizeHandler *ah = get_auth_authorize_handler(con->get_peer_type(),
+							auth_method);
+  if (!ah) {
+    lderr(cct) << __func__ << " no AuthAuthorizeHandler found for auth method "
+	       << auth_method << dendl;
+    return -EOPNOTSUPP;
+  }
+
+  auto ac = &auth_meta->authorizer_challenge;
+  if (auth_meta->skip_authorizer_challenge) {
+    ldout(cct, 10) << __func__ << " skipping challenge on " << con << dendl;
+    ac = nullptr;
+  }
+
+  bool was_challenge = (bool)auth_meta->authorizer_challenge;
+  bool isvalid = ah->verify_authorizer(
+    cct,
+    rotating_secrets.get(),
+    payload,
+    auth_meta->get_connection_secret_length(),
+    reply,
+    &con->peer_name,
+    &con->peer_global_id,
+    &con->peer_caps_info,
+    &auth_meta->session_key,
+    &auth_meta->connection_secret,
+    ac);
+  if (isvalid) {
+    handle_authentication_dispatcher->ms_handle_authentication(con);
+    return 1;
+  }
+  if (!more && !was_challenge && auth_meta->authorizer_challenge) {
+    ldout(cct,10) << __func__ << " added challenge on " << con << dendl;
+    return 0;
+  }
+  ldout(cct,10) << __func__ << " bad authorizer on " << con << dendl;
+  return -EACCES;
+}
+
+AuthAuthorizer* MonClient::build_authorizer(int service_id) const {
+  std::lock_guard l(monc_lock);
+  if (auth) {
+    return auth->build_authorizer(service_id);
+  } else {
+    ldout(cct, 0) << __func__ << " for " << ceph_entity_type_name(service_id)
+		  << ", but no auth is available now" << dendl;
+    return nullptr;
+  }
+}
+
+#define dout_subsys ceph_subsys_monc
+#undef dout_prefix
+#define dout_prefix *_dout << "monclient" << (have_session() ? ": " : "(hunting): ")
+
+MonConnection::MonConnection(
+  CephContext *cct, ConnectionRef con, uint64_t global_id,
+  AuthRegistry *ar)
+  : cct(cct), con(con), global_id(global_id), auth_registry(ar)
+{}
+
+MonConnection::~MonConnection()
+{
+  if (con) {
+    con->mark_down();
+    con.reset();
+  }
+}
+
+bool MonConnection::have_session() const
+{
+  return state == State::HAVE_SESSION;
+}
+
+void MonConnection::start(epoch_t epoch,
+			  const EntityName& entity_name)
+{
+  auth_start = ceph_clock_now();
+
+  if (con->get_peer_addr().is_msgr2()) {
+    ldout(cct, 10) << __func__ << " opening mon connection" << dendl;
+    state = State::AUTHENTICATING;
+    con->send_message(new MMonGetMap());
+    return;
+  }
+
+  // restart authentication handshake
+  state = State::NEGOTIATING;
+
+  // send an initial keepalive to ensure our timestamp is valid by the
+  // time we are in an OPENED state (by sequencing this before
+  // authentication).
+  con->send_keepalive();
+
+  auto m = new MAuth;
+  m->protocol = CEPH_AUTH_UNKNOWN;
+  m->monmap_epoch = epoch;
+  __u8 struct_v = 1;
+  encode(struct_v, m->auth_payload);
+  vector<uint32_t> auth_supported;
+  auth_registry->get_supported_methods(con->get_peer_type(), &auth_supported);
+  encode(auth_supported, m->auth_payload);
+  encode(entity_name, m->auth_payload);
+  encode(global_id, m->auth_payload);
+  con->send_message(m);
+}
+
+int MonConnection::get_auth_request(
+  uint32_t *method,
+  std::vector<uint32_t> *preferred_modes,
+  bufferlist *bl,
+  const EntityName& entity_name,
+  uint32_t want_keys,
+  RotatingKeyRing* keyring)
+{
+  // choose method
+  if (auth_method < 0) {
+    vector<uint32_t> as;
+    auth_registry->get_supported_methods(con->get_peer_type(), &as);
+    if (as.empty()) {
+      return -EACCES;
+    }
+    auth_method = as.front();
+  }
+  *method = auth_method;
+  auth_registry->get_supported_modes(con->get_peer_type(), auth_method,
+				     preferred_modes);
+  ldout(cct,10) << __func__ << " method " << *method
+		<< " preferred_modes " << *preferred_modes << dendl;
+  if (preferred_modes->empty()) {
+    return -EACCES;
+  }
+
+  int r = _init_auth(*method, entity_name, want_keys, keyring, true);
+  ceph_assert(r == 0);
+
+  // initial requset includes some boilerplate...
+  encode((char)AUTH_MODE_MON, *bl);
+  encode(entity_name, *bl);
+  encode(global_id, *bl);
+
+  // and (maybe) some method-specific initial payload
+  auth->build_initial_request(bl);
+
+  return 0;
+}
+
+int MonConnection::handle_auth_reply_more(
+  AuthConnectionMeta *auth_meta,
+  const bufferlist& bl,
+  bufferlist *reply)
+{
+  ldout(cct, 10) << __func__ << " payload " << bl.length() << dendl;
+  ldout(cct, 30) << __func__ << " got\n";
+  bl.hexdump(*_dout);
+  *_dout << dendl;
+
+  auto p = bl.cbegin();
+  ldout(cct, 10) << __func__ << " payload_len " << bl.length() << dendl;
+  int r = auth->handle_response(0, p, &auth_meta->session_key,
+				&auth_meta->connection_secret);
+  if (r == -EAGAIN) {
+    auth->prepare_build_request();
+    auth->build_request(*reply);
+    ldout(cct, 10) << __func__ << " responding with " << reply->length()
+		   << " bytes" << dendl;
+    r = 0;
+  } else if (r < 0) {
+    lderr(cct) << __func__ << " handle_response returned " << r << dendl;
+  } else {
+    ldout(cct, 10) << __func__ << " authenticated!" << dendl;
+    // FIXME
+    ceph_abort(cct, "write me");
+  }
+  return r;
+}
+
+int MonConnection::handle_auth_done(
+  AuthConnectionMeta *auth_meta,
+  uint64_t new_global_id,
+  const bufferlist& bl,
+  CryptoKey *session_key,
+  std::string *connection_secret)
+{
+  ldout(cct,10) << __func__ << " global_id " << new_global_id
+		<< " payload " << bl.length()
+		<< dendl;
+  global_id = new_global_id;
+  auth->set_global_id(global_id);
+  auto p = bl.begin();
+  int auth_err = auth->handle_response(0, p, &auth_meta->session_key,
+				       &auth_meta->connection_secret);
+  if (auth_err >= 0) {
+    state = State::HAVE_SESSION;
+  }
+  con->set_last_keepalive_ack(auth_start);
+  return auth_err;
+}
+
+int MonConnection::handle_auth_bad_method(
+  uint32_t old_auth_method,
+  int result,
+  const std::vector<uint32_t>& allowed_methods,
+  const std::vector<uint32_t>& allowed_modes)
+{
+  ldout(cct,10) << __func__ << " old_auth_method " << old_auth_method
+		<< " result " << cpp_strerror(result)
+		<< " allowed_methods " << allowed_methods << dendl;
+  vector<uint32_t> auth_supported;
+  auth_registry->get_supported_methods(con->get_peer_type(), &auth_supported);
+  auto p = std::find(auth_supported.begin(), auth_supported.end(),
+		     old_auth_method);
+  assert(p != auth_supported.end());
+  p = std::find_first_of(std::next(p), auth_supported.end(),
+			 allowed_methods.begin(), allowed_methods.end());
+  if (p == auth_supported.end()) {
+    lderr(cct) << __func__ << " server allowed_methods " << allowed_methods
+	       << " but i only support " << auth_supported << dendl;
+    return -EACCES;
+  }
+  auth_method = *p;
+  ldout(cct,10) << __func__ << " will try " << auth_method << " next" << dendl;
+  return 0;
+}
+
+int MonConnection::handle_auth(MAuthReply* m,
+			       const EntityName& entity_name,
+			       uint32_t want_keys,
+			       RotatingKeyRing* keyring)
+{
+  if (state == State::NEGOTIATING) {
+    int r = _negotiate(m, entity_name, want_keys, keyring);
+    if (r) {
+      return r;
+    }
+    state = State::AUTHENTICATING;
+  }
+  int r = authenticate(m);
+  if (!r) {
+    state = State::HAVE_SESSION;
+  }
+  return r;
+}
+
+int MonConnection::_negotiate(MAuthReply *m,
+			      const EntityName& entity_name,
+			      uint32_t want_keys,
+			      RotatingKeyRing* keyring)
+{
+  int r = _init_auth(m->protocol, entity_name, want_keys, keyring, false);
+  if (r == -ENOTSUP) {
+    if (m->result == -ENOTSUP) {
+      ldout(cct, 10) << "none of our auth protocols are supported by the server"
+		     << dendl;
+    }
+    return m->result;
+  }
+  return r;
+}
+
+int MonConnection::_init_auth(
+  uint32_t method,
+  const EntityName& entity_name,
+  uint32_t want_keys,
+  RotatingKeyRing* keyring,
+  bool msgr2)
+{
+  ldout(cct, 10) << __func__ << " method " << method << dendl;
+  if (auth && auth->get_protocol() == (int)method) {
+    ldout(cct, 10) << __func__ << " already have auth, reseting" << dendl;
+    auth->reset();
+    return 0;
+  }
+
+  ldout(cct, 10) << __func__ << " creating new auth" << dendl;
+  auth.reset(AuthClientHandler::create(cct, method, keyring));
+  if (!auth) {
+    ldout(cct, 10) << " no handler for protocol " << method << dendl;
+    return -ENOTSUP;
+  }
+
+  // do not request MGR key unless the mon has the SERVER_KRAKEN
+  // feature.  otherwise it will give us an auth error.  note that
+  // we have to use the FEATUREMASK because pre-jewel the kraken
+  // feature bit was used for something else.
+  if (!msgr2 &&
+      (want_keys & CEPH_ENTITY_TYPE_MGR) &&
+      !(con->has_features(CEPH_FEATUREMASK_SERVER_KRAKEN))) {
+    ldout(cct, 1) << __func__
+		  << " not requesting MGR keys from pre-kraken monitor"
+		  << dendl;
+    want_keys &= ~CEPH_ENTITY_TYPE_MGR;
+  }
+  auth->set_want_keys(want_keys);
+  auth->init(entity_name);
+  auth->set_global_id(global_id);
+  return 0;
+}
+
+int MonConnection::authenticate(MAuthReply *m)
+{
+  ceph_assert(auth);
+  if (!m->global_id) {
+    ldout(cct, 1) << "peer sent an invalid global_id" << dendl;
+  }
+  if (m->global_id != global_id) {
+    // it's a new session
+    auth->reset();
+    global_id = m->global_id;
+    auth->set_global_id(global_id);
+    ldout(cct, 10) << "my global_id is " << m->global_id << dendl;
+  }
+  auto p = m->result_bl.cbegin();
+  int ret = auth->handle_response(m->result, p, nullptr, nullptr);
+  if (ret == -EAGAIN) {
+    auto ma = new MAuth;
+    ma->protocol = auth->get_protocol();
+    auth->prepare_build_request();
+    auth->build_request(ma->auth_payload);
+    con->send_message(ma);
+  }
+  return ret;
+}
+
+void MonClient::register_config_callback(md_config_t::config_callback fn) {
+  ceph_assert(!config_cb);
+  config_cb = fn;
+}
+
+md_config_t::config_callback MonClient::get_config_callback() {
+  return config_cb;
+}