summaryrefslogtreecommitdiffstats
path: root/src/mon/MonClient.cc
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-27 18:24:20 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-27 18:24:20 +0000
commit483eb2f56657e8e7f419ab1a4fab8dce9ade8609 (patch)
treee5d88d25d870d5dedacb6bbdbe2a966086a0a5cf /src/mon/MonClient.cc
parentInitial commit. (diff)
downloadceph-upstream.tar.xz
ceph-upstream.zip
Adding upstream version 14.2.21.upstream/14.2.21upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r--src/mon/MonClient.cc1760
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;
+}