1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
8 * This is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License version 2.1, as published by the Free Software
11 * Foundation. See file COPYING.
17 #include "include/scope_guard.h"
19 #include "messages/MMonGetMap.h"
20 #include "messages/MMonGetVersion.h"
21 #include "messages/MMonGetVersionReply.h"
22 #include "messages/MMonMap.h"
23 #include "messages/MAuth.h"
24 #include "messages/MLogAck.h"
25 #include "messages/MAuthReply.h"
26 #include "messages/MMonCommand.h"
27 #include "messages/MMonCommandAck.h"
28 #include "messages/MPing.h"
30 #include "messages/MMonSubscribe.h"
31 #include "messages/MMonSubscribeAck.h"
32 #include "common/errno.h"
33 #include "common/LogClient.h"
35 #include "MonClient.h"
38 #include "auth/Auth.h"
39 #include "auth/KeyRing.h"
40 #include "auth/AuthClientHandler.h"
41 #include "auth/AuthMethodList.h"
42 #include "auth/RotatingKeyRing.h"
44 #define dout_subsys ceph_subsys_monc
46 #define dout_prefix *_dout << "monclient" << (_hunting() ? "(hunting)":"") << ": "
48 MonClient::MonClient(CephContext *cct_) :
51 monc_lock("MonClient::monc_lock"),
52 timer(cct_, monc_lock),
55 no_keyring_disabled_cephx(false),
57 more_log_pending(false),
59 had_a_connection(false),
60 reopen_interval_multiplier(
61 cct_->_conf->get_val<double>("mon_client_hunt_interval_min_multiple")),
62 last_mon_command_tid(0),
67 MonClient::~MonClient()
71 int MonClient::build_initial_monmap()
73 ldout(cct, 10) << __func__ << dendl;
74 return monmap.build_initial(cct, cerr);
77 int MonClient::get_monmap()
79 ldout(cct, 10) << __func__ << dendl;
80 Mutex::Locker l(monc_lock);
82 _sub_want("monmap", 0, 0);
87 map_cond.Wait(monc_lock);
89 ldout(cct, 10) << __func__ << " done" << dendl;
93 int MonClient::get_monmap_privately()
95 ldout(cct, 10) << __func__ << dendl;
96 Mutex::Locker l(monc_lock);
98 bool temp_msgr = false;
99 Messenger* smessenger = NULL;
101 messenger = smessenger = Messenger::create_client_messenger(cct, "temp_mon_client");
102 if (NULL == messenger) {
105 messenger->add_dispatcher_head(this);
112 ldout(cct, 10) << "have " << monmap.epoch << " fsid " << monmap.fsid << dendl;
114 std::random_device rd;
115 std::mt19937 rng(rd());
116 assert(monmap.size() > 0);
117 std::uniform_int_distribution<unsigned> ranks(0, monmap.size() - 1);
118 while (monmap.fsid.is_zero()) {
119 auto rank = ranks(rng);
120 auto& pending_con = _add_conn(rank, 0);
121 auto con = pending_con.get_con();
122 ldout(cct, 10) << "querying mon." << monmap.get_name(rank) << " "
123 << con->get_peer_addr() << dendl;
124 con->send_message(new MMonGetMap);
130 interval.set_from_double(cct->_conf->mon_client_hunt_interval);
131 map_cond.WaitInterval(monc_lock, interval);
133 if (monmap.fsid.is_zero() && con) {
134 con->mark_down(); // nope, clean that connection up
139 pending_cons.clear();
141 messenger->shutdown();
149 pending_cons.clear();
151 if (!monmap.fsid.is_zero())
158 * Ping the monitor with id @p mon_id and set the resulting reply in
159 * the provided @p result_reply, if this last parameter is not NULL.
161 * So that we don't rely on the MonClient's default messenger, set up
162 * during connect(), we create our own messenger to comunicate with the
163 * specified monitor. This is advantageous in the following ways:
165 * - Isolate the ping procedure from the rest of the MonClient's operations,
166 * allowing us to not acquire or manage the big monc_lock, thus not
167 * having to block waiting for some other operation to finish before we
169 * * for instance, we can ping mon.FOO even if we are currently hunting
170 * or blocked waiting for auth to complete with mon.BAR.
172 * - Ping a monitor prior to establishing a connection (using connect())
173 * and properly establish the MonClient's messenger. This frees us
174 * from dealing with the complex foo that happens in connect().
176 * We also don't rely on MonClient as a dispatcher for this messenger,
177 * unlike what happens with the MonClient's default messenger. This allows
178 * us to sandbox the whole ping, having it much as a separate entity in
179 * the MonClient class, considerably simplifying the handling and dispatching
180 * of messages without needing to consider monc_lock.
182 * Current drawback is that we will establish a messenger for each ping
183 * we want to issue, instead of keeping a single messenger instance that
184 * would be used for all pings.
186 int MonClient::ping_monitor(const string &mon_id, string *result_reply)
188 ldout(cct, 10) << __func__ << dendl;
191 if (monmap.contains("noname-"+mon_id)) {
192 new_mon_id = "noname-"+mon_id;
197 if (new_mon_id.empty()) {
198 ldout(cct, 10) << __func__ << " specified mon id is empty!" << dendl;
200 } else if (!monmap.contains(new_mon_id)) {
201 ldout(cct, 10) << __func__ << " no such monitor 'mon." << new_mon_id << "'"
206 MonClientPinger *pinger = new MonClientPinger(cct, result_reply);
208 Messenger *smsgr = Messenger::create_client_messenger(cct, "temp_ping_client");
209 smsgr->add_dispatcher_head(pinger);
212 ConnectionRef con = smsgr->get_connection(monmap.get_inst(new_mon_id));
213 ldout(cct, 10) << __func__ << " ping mon." << new_mon_id
214 << " " << con->get_peer_addr() << dendl;
215 con->send_message(new MPing);
218 int ret = pinger->wait_for_reply(cct->_conf->client_mount_timeout);
220 ldout(cct,10) << __func__ << " got ping reply" << dendl;
224 pinger->lock.Unlock();
234 bool MonClient::ms_dispatch(Message *m)
236 if (my_addr == entity_addr_t())
237 my_addr = messenger->get_myaddr();
239 // we only care about these message types
240 switch (m->get_type()) {
241 case CEPH_MSG_MON_MAP:
242 case CEPH_MSG_AUTH_REPLY:
243 case CEPH_MSG_MON_SUBSCRIBE_ACK:
244 case CEPH_MSG_MON_GET_VERSION_REPLY:
245 case MSG_MON_COMMAND_ACK:
252 Mutex::Locker lock(monc_lock);
255 auto pending_con = pending_cons.find(m->get_source_addr());
256 if (pending_con == pending_cons.end() ||
257 pending_con->second.get_con() != m->get_connection()) {
258 // ignore any messages outside hunting sessions
259 ldout(cct, 10) << "discarding stray monitor message " << *m << dendl;
263 } else if (!active_con || active_con->get_con() != m->get_connection()) {
264 // ignore any messages outside our session(s)
265 ldout(cct, 10) << "discarding stray monitor message " << *m << dendl;
270 switch (m->get_type()) {
271 case CEPH_MSG_MON_MAP:
272 handle_monmap(static_cast<MMonMap*>(m));
273 if (passthrough_monmap) {
279 case CEPH_MSG_AUTH_REPLY:
280 handle_auth(static_cast<MAuthReply*>(m));
282 case CEPH_MSG_MON_SUBSCRIBE_ACK:
283 handle_subscribe_ack(static_cast<MMonSubscribeAck*>(m));
285 case CEPH_MSG_MON_GET_VERSION_REPLY:
286 handle_get_version_reply(static_cast<MMonGetVersionReply*>(m));
288 case MSG_MON_COMMAND_ACK:
289 handle_mon_command_ack(static_cast<MMonCommandAck*>(m));
293 log_client->handle_log_ack(static_cast<MLogAck*>(m));
295 if (more_log_pending) {
306 void MonClient::send_log(bool flush)
309 Message *lm = log_client->get_mon_log_message(flush);
311 _send_mon_message(lm);
312 more_log_pending = log_client->are_pending();
316 void MonClient::flush_log()
318 Mutex::Locker l(monc_lock);
322 /* Unlike all the other message-handling functions, we don't put away a reference
323 * because we want to support MMonMap passthrough to other Dispatchers. */
324 void MonClient::handle_monmap(MMonMap *m)
326 ldout(cct, 10) << __func__ << " " << *m << dendl;
327 auto peer = m->get_source_addr();
328 string cur_mon = monmap.get_name(peer);
330 bufferlist::iterator p = m->monmapbl.begin();
333 ldout(cct, 10) << " got monmap " << monmap.epoch
334 << ", mon." << cur_mon << " is now rank " << monmap.get_rank(cur_mon)
336 ldout(cct, 10) << "dump:\n";
337 monmap.print(*_dout);
340 _sub_got("monmap", monmap.get_epoch());
342 if (!monmap.get_addr_name(peer, cur_mon)) {
343 ldout(cct, 10) << "mon." << cur_mon << " went away" << dendl;
344 // can't find the mon we were talking to (above)
352 // ----------------------
354 int MonClient::init()
356 ldout(cct, 10) << __func__ << dendl;
358 messenger->add_dispatcher_head(this);
360 entity_name = cct->_conf->name;
362 Mutex::Locker l(monc_lock);
365 if (!cct->_conf->auth_supported.empty())
366 method = cct->_conf->auth_supported;
367 else if (entity_name.get_type() == CEPH_ENTITY_TYPE_OSD ||
368 entity_name.get_type() == CEPH_ENTITY_TYPE_MDS ||
369 entity_name.get_type() == CEPH_ENTITY_TYPE_MON)
370 method = cct->_conf->auth_cluster_required;
372 method = cct->_conf->auth_client_required;
373 auth_supported.reset(new AuthMethodList(cct, method));
374 ldout(cct, 10) << "auth_supported " << auth_supported->get_supported_set() << " method " << method << dendl;
377 keyring.reset(new KeyRing); // initializing keyring anyway
379 if (auth_supported->is_supported_auth(CEPH_AUTH_CEPHX)) {
380 r = keyring->from_ceph_context(cct);
382 auth_supported->remove_supported_auth(CEPH_AUTH_CEPHX);
383 if (!auth_supported->get_supported_set().empty()) {
385 no_keyring_disabled_cephx = true;
387 lderr(cct) << "ERROR: missing keyring, cannot use cephx for authentication" << dendl;
396 rotating_secrets.reset(
397 new RotatingKeyRing(cct, cct->get_module_type(), keyring.get()));
408 void MonClient::shutdown()
410 ldout(cct, 10) << __func__ << dendl;
412 while (!version_requests.empty()) {
413 version_requests.begin()->second->context->complete(-ECANCELED);
414 ldout(cct, 20) << __func__ << " canceling and discarding version request "
415 << version_requests.begin()->second << dendl;
416 delete version_requests.begin()->second;
417 version_requests.erase(version_requests.begin());
419 while (!mon_commands.empty()) {
420 auto tid = mon_commands.begin()->first;
421 _cancel_mon_command(tid);
423 while (!waiting_for_session.empty()) {
424 ldout(cct, 20) << __func__ << " discarding pending message " << *waiting_for_session.front() << dendl;
425 waiting_for_session.front()->put();
426 waiting_for_session.pop_front();
430 pending_cons.clear();
436 finisher.wait_for_empty();
445 int MonClient::authenticate(double timeout)
447 Mutex::Locker lock(monc_lock);
450 ldout(cct, 5) << "already authenticated" << dendl;
454 _sub_want("monmap", monmap.get_epoch() ? monmap.get_epoch() + 1 : 0, 0);
458 utime_t until = ceph_clock_now();
461 ldout(cct, 10) << "authenticate will time out at " << until << dendl;
462 while (!active_con && !authenticate_err) {
464 int r = auth_cond.WaitUntil(monc_lock, until);
465 if (r == ETIMEDOUT) {
466 ldout(cct, 0) << "authenticate timed out after " << timeout << dendl;
467 authenticate_err = -r;
470 auth_cond.Wait(monc_lock);
475 ldout(cct, 5) << __func__ << " success, global_id "
476 << active_con->get_global_id() << dendl;
477 // active_con should not have been set if there was an error
478 assert(authenticate_err == 0);
479 authenticated = true;
482 if (authenticate_err < 0 && no_keyring_disabled_cephx) {
483 lderr(cct) << __func__ << " NOTE: no keyring found; disabled cephx authentication" << dendl;
486 return authenticate_err;
489 void MonClient::handle_auth(MAuthReply *m)
491 assert(monc_lock.is_locked());
493 std::swap(active_con->get_auth(), auth);
494 int ret = active_con->authenticate(m);
496 std::swap(auth, active_con->get_auth());
497 if (global_id != active_con->get_global_id()) {
498 lderr(cct) << __func__ << " peer assigned me a different global_id: "
499 << active_con->get_global_id() << dendl;
501 if (ret != -EAGAIN) {
508 auto found = pending_cons.find(m->get_source_addr());
509 assert(found != pending_cons.end());
510 int auth_err = found->second.handle_auth(m, entity_name, want_keys,
511 rotating_secrets.get());
513 if (auth_err == -EAGAIN) {
517 pending_cons.erase(found);
518 if (!pending_cons.empty()) {
519 // keep trying with pending connections
522 // the last try just failed, give up.
524 auto& mc = found->second;
525 assert(mc.have_session());
526 active_con.reset(new MonConnection(std::move(mc)));
527 pending_cons.clear();
533 last_rotating_renew_sent = utime_t();
534 while (!waiting_for_session.empty()) {
535 _send_mon_message(waiting_for_session.front());
536 waiting_for_session.pop_front();
538 _resend_mon_commands();
541 std::swap(auth, active_con->get_auth());
542 global_id = active_con->get_global_id();
545 _finish_auth(auth_err);
547 Context *cb = nullptr;
548 if (session_established_context) {
549 cb = session_established_context.release();
559 void MonClient::_finish_auth(int auth_err)
561 authenticate_err = auth_err;
562 // _resend_mon_commands() could _reopen_session() if the connected mon is not
563 // the one the MonCommand is targeting.
564 if (!auth_err && active_con) {
566 _check_auth_tickets();
568 auth_cond.SignalAll();
573 void MonClient::_send_mon_message(Message *m)
575 assert(monc_lock.is_locked());
577 auto cur_con = active_con->get_con();
578 ldout(cct, 10) << "_send_mon_message to mon."
579 << monmap.get_name(cur_con->get_peer_addr())
580 << " at " << cur_con->get_peer_addr() << dendl;
581 cur_con->send_message(m);
583 waiting_for_session.push_back(m);
587 void MonClient::_reopen_session(int rank)
589 assert(monc_lock.is_locked());
590 ldout(cct, 10) << __func__ << " rank " << rank << dendl;
593 pending_cons.clear();
598 _add_conn(rank, global_id);
600 _add_conns(global_id);
603 // throw out old queued messages
604 while (!waiting_for_session.empty()) {
605 waiting_for_session.front()->put();
606 waiting_for_session.pop_front();
609 // throw out version check requests
610 while (!version_requests.empty()) {
611 finisher.queue(version_requests.begin()->second->context, -EAGAIN);
612 delete version_requests.begin()->second;
613 version_requests.erase(version_requests.begin());
616 for (auto& c : pending_cons) {
617 c.second.start(monmap.get_epoch(), entity_name, *auth_supported);
620 for (map<string,ceph_mon_subscribe_item>::iterator p = sub_sent.begin();
623 if (sub_new.count(p->first) == 0)
624 sub_new[p->first] = p->second;
626 if (!sub_new.empty())
630 MonConnection& MonClient::_add_conn(unsigned rank, uint64_t global_id)
632 auto peer = monmap.get_addr(rank);
633 auto conn = messenger->get_connection(monmap.get_inst(rank));
634 MonConnection mc(cct, conn, global_id);
635 auto inserted = pending_cons.insert(make_pair(peer, move(mc)));
636 ldout(cct, 10) << "picked mon." << monmap.get_name(rank)
638 << " addr " << conn->get_peer_addr()
640 return inserted.first->second;
643 void MonClient::_add_conns(uint64_t global_id)
645 uint16_t min_priority = std::numeric_limits<uint16_t>::max();
646 for (const auto& m : monmap.mon_info) {
647 if (m.second.priority < min_priority) {
648 min_priority = m.second.priority;
651 vector<unsigned> ranks;
652 for (const auto& m : monmap.mon_info) {
653 if (m.second.priority == min_priority) {
654 ranks.push_back(monmap.get_rank(m.first));
657 std::random_device rd;
658 std::mt19937 rng(rd());
659 std::shuffle(ranks.begin(), ranks.end(), rng);
660 unsigned n = cct->_conf->mon_client_hunt_parallel;
661 if (n == 0 || n > ranks.size()) {
664 for (unsigned i = 0; i < n; i++) {
665 _add_conn(ranks[i], global_id);
669 bool MonClient::ms_handle_reset(Connection *con)
671 Mutex::Locker lock(monc_lock);
673 if (con->get_peer_type() != CEPH_ENTITY_TYPE_MON)
677 if (pending_cons.count(con->get_peer_addr())) {
678 ldout(cct, 10) << __func__ << " hunted mon " << con->get_peer_addr() << dendl;
680 ldout(cct, 10) << __func__ << " stray mon " << con->get_peer_addr() << dendl;
684 if (active_con && con == active_con->get_con()) {
685 ldout(cct, 10) << __func__ << " current mon " << con->get_peer_addr() << dendl;
689 ldout(cct, 10) << "ms_handle_reset stray mon " << con->get_peer_addr() << dendl;
695 bool MonClient::_opened() const
697 assert(monc_lock.is_locked());
698 return active_con || _hunting();
701 bool MonClient::_hunting() const
703 return !pending_cons.empty();
706 void MonClient::_start_hunting()
709 // adjust timeouts if necessary
710 if (!had_a_connection)
712 reopen_interval_multiplier *= cct->_conf->mon_client_hunt_interval_backoff;
713 if (reopen_interval_multiplier >
714 cct->_conf->mon_client_hunt_interval_max_multiple) {
715 reopen_interval_multiplier =
716 cct->_conf->mon_client_hunt_interval_max_multiple;
720 void MonClient::_finish_hunting()
722 assert(monc_lock.is_locked());
723 // the pending conns have been cleaned.
726 auto con = active_con->get_con();
727 ldout(cct, 1) << "found mon."
728 << monmap.get_name(con->get_peer_addr())
731 ldout(cct, 1) << "no mon sessions established" << dendl;
734 had_a_connection = true;
738 void MonClient::tick()
740 ldout(cct, 10) << __func__ << dendl;
742 auto reschedule_tick = make_scope_guard([this] {
746 _check_auth_tickets();
749 ldout(cct, 1) << "continuing hunt" << dendl;
750 return _reopen_session();
751 } else if (active_con) {
752 // just renew as needed
753 utime_t now = ceph_clock_now();
754 auto cur_con = active_con->get_con();
755 if (!cur_con->has_feature(CEPH_FEATURE_MON_STATEFUL_SUB)) {
756 ldout(cct, 10) << "renew subs? (now: " << now
757 << "; renew after: " << sub_renew_after << ") -- "
758 << (now > sub_renew_after ? "yes" : "no")
760 if (now > sub_renew_after)
764 cur_con->send_keepalive();
766 if (cct->_conf->mon_client_ping_timeout > 0 &&
767 cur_con->has_feature(CEPH_FEATURE_MSGR_KEEPALIVE2)) {
768 utime_t lk = cur_con->get_last_keepalive_ack();
769 utime_t interval = now - lk;
770 if (interval > cct->_conf->mon_client_ping_timeout) {
771 ldout(cct, 1) << "no keepalive since " << lk << " (" << interval
772 << " seconds), reconnecting" << dendl;
773 return _reopen_session();
782 void MonClient::_un_backoff()
784 // un-backoff our reconnect interval
785 reopen_interval_multiplier = std::max(
786 cct->_conf->get_val<double>("mon_client_hunt_interval_min_multiple"),
787 reopen_interval_multiplier /
788 cct->_conf->get_val<double>("mon_client_hunt_interval_backoff"));
789 ldout(cct, 20) << __func__ << " reopen_interval_multipler now "
790 << reopen_interval_multiplier << dendl;
793 void MonClient::schedule_tick()
795 struct C_Tick : public Context {
797 explicit C_Tick(MonClient *m) : monc(m) {}
798 void finish(int r) override {
804 timer.add_event_after(cct->_conf->mon_client_hunt_interval
805 * reopen_interval_multiplier,
808 timer.add_event_after(cct->_conf->mon_client_ping_interval, new C_Tick(this));
813 void MonClient::_renew_subs()
815 assert(monc_lock.is_locked());
816 if (sub_new.empty()) {
817 ldout(cct, 10) << __func__ << " - empty" << dendl;
821 ldout(cct, 10) << __func__ << dendl;
825 if (sub_renew_sent == utime_t())
826 sub_renew_sent = ceph_clock_now();
828 MMonSubscribe *m = new MMonSubscribe;
830 _send_mon_message(m);
832 // update sub_sent with sub_new
833 sub_new.insert(sub_sent.begin(), sub_sent.end());
834 std::swap(sub_new, sub_sent);
839 void MonClient::handle_subscribe_ack(MMonSubscribeAck *m)
841 if (sub_renew_sent != utime_t()) {
842 // NOTE: this is only needed for legacy (infernalis or older)
844 sub_renew_after = sub_renew_sent;
845 sub_renew_after += m->interval / 2.0;
846 ldout(cct, 10) << __func__ << " sent " << sub_renew_sent << " renew after " << sub_renew_after << dendl;
847 sub_renew_sent = utime_t();
849 ldout(cct, 10) << __func__ << " sent " << sub_renew_sent << ", ignoring" << dendl;
855 int MonClient::_check_auth_tickets()
857 assert(monc_lock.is_locked());
858 if (active_con && auth) {
859 if (auth->need_tickets()) {
860 ldout(cct, 10) << __func__ << " getting new tickets!" << dendl;
861 MAuth *m = new MAuth;
862 m->protocol = auth->get_protocol();
863 auth->prepare_build_request();
864 auth->build_request(m->auth_payload);
865 _send_mon_message(m);
868 _check_auth_rotating();
873 int MonClient::_check_auth_rotating()
875 assert(monc_lock.is_locked());
876 if (!rotating_secrets ||
877 !auth_principal_needs_rotating_keys(entity_name)) {
878 ldout(cct, 20) << "_check_auth_rotating not needed by " << entity_name << dendl;
882 if (!active_con || !auth) {
883 ldout(cct, 10) << "_check_auth_rotating waiting for auth session" << dendl;
887 utime_t now = ceph_clock_now();
888 utime_t cutoff = now;
889 cutoff -= MIN(30.0, cct->_conf->auth_service_ticket_ttl / 4.0);
890 utime_t issued_at_lower_bound = now;
891 issued_at_lower_bound -= cct->_conf->auth_service_ticket_ttl;
892 if (!rotating_secrets->need_new_secrets(cutoff)) {
893 ldout(cct, 10) << "_check_auth_rotating have uptodate secrets (they expire after " << cutoff << ")" << dendl;
894 rotating_secrets->dump_rotating();
898 ldout(cct, 10) << "_check_auth_rotating renewing rotating keys (they expired before " << cutoff << ")" << dendl;
899 if (!rotating_secrets->need_new_secrets() &&
900 rotating_secrets->need_new_secrets(issued_at_lower_bound)) {
901 // the key has expired before it has been issued?
902 lderr(cct) << __func__ << " possible clock skew, rotating keys expired way too early"
903 << " (before " << issued_at_lower_bound << ")" << dendl;
905 if ((now > last_rotating_renew_sent) &&
906 double(now - last_rotating_renew_sent) < 1) {
907 ldout(cct, 10) << __func__ << " called too often (last: "
908 << last_rotating_renew_sent << "), skipping refresh" << dendl;
911 MAuth *m = new MAuth;
912 m->protocol = auth->get_protocol();
913 if (auth->build_rotating_request(m->auth_payload)) {
914 last_rotating_renew_sent = now;
915 _send_mon_message(m);
922 int MonClient::wait_auth_rotating(double timeout)
924 Mutex::Locker l(monc_lock);
925 utime_t now = ceph_clock_now();
929 // Must be initialized
930 assert(auth != nullptr);
932 if (auth->get_protocol() == CEPH_AUTH_NONE)
935 if (!rotating_secrets)
938 while (auth_principal_needs_rotating_keys(entity_name) &&
939 rotating_secrets->need_new_secrets(now)) {
941 ldout(cct, 0) << __func__ << " timed out after " << timeout << dendl;
944 ldout(cct, 10) << __func__ << " waiting (until " << until << ")" << dendl;
945 auth_cond.WaitUntil(monc_lock, until);
946 now = ceph_clock_now();
948 ldout(cct, 10) << __func__ << " done" << dendl;
954 void MonClient::_send_command(MonCommand *r)
958 peer = active_con->get_con()->get_peer_addr();
961 if (r->target_rank >= 0 &&
962 r->target_rank != monmap.get_rank(peer)) {
963 ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd
964 << " wants rank " << r->target_rank
965 << ", reopening session"
967 if (r->target_rank >= (int)monmap.size()) {
968 ldout(cct, 10) << " target " << r->target_rank << " >= max mon " << monmap.size() << dendl;
969 _finish_command(r, -ENOENT, "mon rank dne");
972 _reopen_session(r->target_rank);
976 if (r->target_name.length() &&
977 r->target_name != monmap.get_name(peer)) {
978 ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd
979 << " wants mon " << r->target_name
980 << ", reopening session"
982 if (!monmap.contains(r->target_name)) {
983 ldout(cct, 10) << " target " << r->target_name << " not present in monmap" << dendl;
984 _finish_command(r, -ENOENT, "mon dne");
987 _reopen_session(monmap.get_rank(r->target_name));
991 ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
992 MMonCommand *m = new MMonCommand(monmap.fsid);
995 m->set_data(r->inbl);
996 _send_mon_message(m);
1000 void MonClient::_resend_mon_commands()
1002 // resend any requests
1003 for (map<uint64_t,MonCommand*>::iterator p = mon_commands.begin();
1004 p != mon_commands.end();
1006 _send_command(p->second);
1010 void MonClient::handle_mon_command_ack(MMonCommandAck *ack)
1012 MonCommand *r = NULL;
1013 uint64_t tid = ack->get_tid();
1015 if (tid == 0 && !mon_commands.empty()) {
1016 r = mon_commands.begin()->second;
1017 ldout(cct, 10) << __func__ << " has tid 0, assuming it is " << r->tid << dendl;
1019 map<uint64_t,MonCommand*>::iterator p = mon_commands.find(tid);
1020 if (p == mon_commands.end()) {
1021 ldout(cct, 10) << __func__ << " " << ack->get_tid() << " not found" << dendl;
1028 ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
1030 r->poutbl->claim(ack->get_data());
1031 _finish_command(r, ack->r, ack->rs);
1035 int MonClient::_cancel_mon_command(uint64_t tid)
1037 assert(monc_lock.is_locked());
1039 map<ceph_tid_t, MonCommand*>::iterator it = mon_commands.find(tid);
1040 if (it == mon_commands.end()) {
1041 ldout(cct, 10) << __func__ << " tid " << tid << " dne" << dendl;
1045 ldout(cct, 10) << __func__ << " tid " << tid << dendl;
1047 MonCommand *cmd = it->second;
1048 _finish_command(cmd, -ETIMEDOUT, "");
1052 void MonClient::_finish_command(MonCommand *r, int ret, string rs)
1054 ldout(cct, 10) << __func__ << " " << r->tid << " = " << ret << " " << rs << dendl;
1060 finisher.queue(r->onfinish, ret);
1061 mon_commands.erase(r->tid);
1065 void MonClient::start_mon_command(const vector<string>& cmd,
1066 const bufferlist& inbl,
1067 bufferlist *outbl, string *outs,
1070 Mutex::Locker l(monc_lock);
1071 MonCommand *r = new MonCommand(++last_mon_command_tid);
1076 r->onfinish = onfinish;
1077 if (cct->_conf->rados_mon_op_timeout > 0) {
1078 class C_CancelMonCommand : public Context
1083 C_CancelMonCommand(uint64_t tid, MonClient *monc) : tid(tid), monc(monc) {}
1084 void finish(int r) override {
1085 monc->_cancel_mon_command(tid);
1088 r->ontimeout = new C_CancelMonCommand(r->tid, this);
1089 timer.add_event_after(cct->_conf->rados_mon_op_timeout, r->ontimeout);
1091 mon_commands[r->tid] = r;
1095 void MonClient::start_mon_command(const string &mon_name,
1096 const vector<string>& cmd,
1097 const bufferlist& inbl,
1098 bufferlist *outbl, string *outs,
1101 Mutex::Locker l(monc_lock);
1102 MonCommand *r = new MonCommand(++last_mon_command_tid);
1103 r->target_name = mon_name;
1108 r->onfinish = onfinish;
1109 mon_commands[r->tid] = r;
1113 void MonClient::start_mon_command(int rank,
1114 const vector<string>& cmd,
1115 const bufferlist& inbl,
1116 bufferlist *outbl, string *outs,
1119 Mutex::Locker l(monc_lock);
1120 MonCommand *r = new MonCommand(++last_mon_command_tid);
1121 r->target_rank = rank;
1126 r->onfinish = onfinish;
1127 mon_commands[r->tid] = r;
1133 void MonClient::get_version(string map, version_t *newest, version_t *oldest, Context *onfinish)
1135 version_req_d *req = new version_req_d(onfinish, newest, oldest);
1136 ldout(cct, 10) << "get_version " << map << " req " << req << dendl;
1137 Mutex::Locker l(monc_lock);
1138 MMonGetVersion *m = new MMonGetVersion();
1140 m->handle = ++version_req_id;
1141 version_requests[m->handle] = req;
1142 _send_mon_message(m);
1145 void MonClient::handle_get_version_reply(MMonGetVersionReply* m)
1147 assert(monc_lock.is_locked());
1148 map<ceph_tid_t, version_req_d*>::iterator iter = version_requests.find(m->handle);
1149 if (iter == version_requests.end()) {
1150 ldout(cct, 0) << __func__ << " version request with handle " << m->handle
1151 << " not found" << dendl;
1153 version_req_d *req = iter->second;
1154 ldout(cct, 10) << __func__ << " finishing " << req << " version " << m->version << dendl;
1155 version_requests.erase(iter);
1157 *req->newest = m->version;
1159 *req->oldest = m->oldest_version;
1160 finisher.queue(req->context, 0);
1166 AuthAuthorizer* MonClient::build_authorizer(int service_id) const {
1167 Mutex::Locker l(monc_lock);
1169 return auth->build_authorizer(service_id);
1171 ldout(cct, 0) << __func__ << " for " << ceph_entity_type_name(service_id)
1172 << ", but no auth is available now" << dendl;
1177 #define dout_subsys ceph_subsys_monc
1179 #define dout_prefix *_dout << "monclient" << (have_session() ? ": " : "(hunting): ")
1181 MonConnection::MonConnection(CephContext *cct, ConnectionRef con, uint64_t global_id)
1182 : cct(cct), con(con), global_id(global_id)
1185 MonConnection::~MonConnection()
1193 bool MonConnection::have_session() const
1195 return state == State::HAVE_SESSION;
1198 void MonConnection::start(epoch_t epoch,
1199 const EntityName& entity_name,
1200 const AuthMethodList& auth_supported)
1202 // restart authentication handshake
1203 state = State::NEGOTIATING;
1205 // send an initial keepalive to ensure our timestamp is valid by the
1206 // time we are in an OPENED state (by sequencing this before
1208 con->send_keepalive();
1212 m->monmap_epoch = epoch;
1214 ::encode(struct_v, m->auth_payload);
1215 ::encode(auth_supported.get_supported_set(), m->auth_payload);
1216 ::encode(entity_name, m->auth_payload);
1217 ::encode(global_id, m->auth_payload);
1218 con->send_message(m);
1221 int MonConnection::handle_auth(MAuthReply* m,
1222 const EntityName& entity_name,
1224 RotatingKeyRing* keyring)
1226 if (state == State::NEGOTIATING) {
1227 int r = _negotiate(m, entity_name, want_keys, keyring);
1231 state = State::AUTHENTICATING;
1233 int r = authenticate(m);
1235 state = State::HAVE_SESSION;
1240 int MonConnection::_negotiate(MAuthReply *m,
1241 const EntityName& entity_name,
1243 RotatingKeyRing* keyring)
1245 if (auth && (int)m->protocol == auth->get_protocol()) {
1246 // good, negotiation completed
1251 auth.reset(get_auth_client_handler(cct, m->protocol, keyring));
1253 ldout(cct, 10) << "no handler for protocol " << m->protocol << dendl;
1254 if (m->result == -ENOTSUP) {
1255 ldout(cct, 10) << "none of our auth protocols are supported by the server"
1261 // do not request MGR key unless the mon has the SERVER_KRAKEN
1262 // feature. otherwise it will give us an auth error. note that
1263 // we have to use the FEATUREMASK because pre-jewel the kraken
1264 // feature bit was used for something else.
1265 if ((want_keys & CEPH_ENTITY_TYPE_MGR) &&
1266 !(m->get_connection()->has_features(CEPH_FEATUREMASK_SERVER_KRAKEN))) {
1267 ldout(cct, 1) << __func__
1268 << " not requesting MGR keys from pre-kraken monitor"
1270 want_keys &= ~CEPH_ENTITY_TYPE_MGR;
1272 auth->set_want_keys(want_keys);
1273 auth->init(entity_name);
1274 auth->set_global_id(global_id);
1278 int MonConnection::authenticate(MAuthReply *m)
1281 if (!m->global_id) {
1282 ldout(cct, 1) << "peer sent an invalid global_id" << dendl;
1284 if (m->global_id != global_id) {
1285 // it's a new session
1287 global_id = m->global_id;
1288 auth->set_global_id(global_id);
1289 ldout(cct, 10) << "my global_id is " << m->global_id << dendl;
1291 auto p = m->result_bl.begin();
1292 int ret = auth->handle_response(m->result, p);
1293 if (ret == -EAGAIN) {
1294 auto ma = new MAuth;
1295 ma->protocol = auth->get_protocol();
1296 auth->prepare_build_request();
1297 auth->build_request(ma->auth_payload);
1298 con->send_message(ma);