ceph/src/mon/MonClient.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   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.
  12  *
  13  */
  14
  15 #include <random>
  16
  17 #include "messages/MMonGetMap.h"
  18 #include "messages/MMonGetVersion.h"
  19 #include "messages/MMonGetVersionReply.h"
  20 #include "messages/MMonMap.h"
  21 #include "messages/MAuth.h"
  22 #include "messages/MLogAck.h"
  23 #include "messages/MAuthReply.h"
  24 #include "messages/MMonCommand.h"
  25 #include "messages/MMonCommandAck.h"
  26 #include "messages/MPing.h"
  27
  28 #include "messages/MMonSubscribe.h"
  29 #include "messages/MMonSubscribeAck.h"
  30 #include "common/errno.h"
  31 #include "common/LogClient.h"
  32
  33 #include "MonClient.h"
  34 #include "MonMap.h"
  35
  36 #include "auth/Auth.h"
  37 #include "auth/KeyRing.h"
  38 #include "auth/AuthClientHandler.h"
  39 #include "auth/AuthMethodList.h"
  40 #include "auth/RotatingKeyRing.h"
  41
  42 #define dout_subsys ceph_subsys_monc
  43 #undef dout_prefix
  44 #define dout_prefix *_dout << "monclient" << (_hunting() ? "(hunting)":"") << ": "
  45
  46 MonClient::MonClient(CephContext *cct_) :
  47   Dispatcher(cct_),
  48   messenger(NULL),
  49   monc_lock("MonClient::monc_lock"),
  50   timer(cct_, monc_lock),
  51   finisher(cct_),
  52   initialized(false),
  53   no_keyring_disabled_cephx(false),
  54   log_client(NULL),
  55   more_log_pending(false),
  56   want_monmap(true),
  57   had_a_connection(false),
  58   reopen_interval_multiplier(1.0),
  59   last_mon_command_tid(0),
  60   version_req_id(0)
  61 {
  62 }
  63
  64 MonClient::~MonClient()
  65 {
  66 }
  67
  68 int MonClient::build_initial_monmap()
  69 {
  70   ldout(cct, 10) << __func__ << dendl;
  71   return monmap.build_initial(cct, cerr);
  72 }
  73
  74 int MonClient::get_monmap()
  75 {
  76   ldout(cct, 10) << __func__ << dendl;
  77   Mutex::Locker l(monc_lock);
  78
  79   _sub_want("monmap", 0, 0);
  80   if (!_opened())
  81     _reopen_session();
  82
  83   while (want_monmap)
  84     map_cond.Wait(monc_lock);
  85
  86   ldout(cct, 10) << __func__ << " done" << dendl;
  87   return 0;
  88 }
  89
  90 int MonClient::get_monmap_privately()
  91 {
  92   ldout(cct, 10) << __func__ << dendl;
  93   Mutex::Locker l(monc_lock);
  94
  95   bool temp_msgr = false;
  96   Messenger* smessenger = NULL;
  97   if (!messenger) {
  98     messenger = smessenger = Messenger::create_client_messenger(cct, "temp_mon_client");
  99     if (NULL == messenger) {
 100         return -1;
 101     }
 102     messenger->add_dispatcher_head(this);
 103     smessenger->start();
 104     temp_msgr = true;
 105   }
 106
 107   int attempt = 10;
 108
 109   ldout(cct, 10) << "have " << monmap.epoch << " fsid " << monmap.fsid << dendl;
 110
 111   std::random_device rd;
 112   std::mt19937 rng(rd());
 113   assert(monmap.size() > 0);
 114   std::uniform_int_distribution<unsigned> ranks(0, monmap.size() - 1);
 115   while (monmap.fsid.is_zero()) {
 116     auto rank = ranks(rng);
 117     auto& pending_con = _add_conn(rank, 0);
 118     auto con = pending_con.get_con();
 119     ldout(cct, 10) << "querying mon." << monmap.get_name(rank) << " "
 120                    << con->get_peer_addr() << dendl;
 121     con->send_message(new MMonGetMap);
 122
 123     if (--attempt == 0)
 124       break;
 125
 126     utime_t interval;
 127     interval.set_from_double(cct->_conf->mon_client_hunt_interval);
 128     map_cond.WaitInterval(monc_lock, interval);
 129
 130     if (monmap.fsid.is_zero() && con) {
 131       con->mark_down();  // nope, clean that connection up
 132     }
 133   }
 134
 135   if (temp_msgr) {
 136     pending_cons.clear();
 137     monc_lock.Unlock();
 138     messenger->shutdown();
 139     if (smessenger)
 140       smessenger->wait();
 141     delete messenger;
 142     messenger = 0;
 143     monc_lock.Lock();
 144   }
 145
 146   pending_cons.clear();
 147
 148   if (!monmap.fsid.is_zero())
 149     return 0;
 150   return -1;
 151 }
 152
 153
 154 /**
 155  * Ping the monitor with id @p mon_id and set the resulting reply in
 156  * the provided @p result_reply, if this last parameter is not NULL.
 157  *
 158  * So that we don't rely on the MonClient's default messenger, set up
 159  * during connect(), we create our own messenger to comunicate with the
 160  * specified monitor.  This is advantageous in the following ways:
 161  *
 162  * - Isolate the ping procedure from the rest of the MonClient's operations,
 163  *   allowing us to not acquire or manage the big monc_lock, thus not
 164  *   having to block waiting for some other operation to finish before we
 165  *   can proceed.
 166  *   * for instance, we can ping mon.FOO even if we are currently hunting
 167  *     or blocked waiting for auth to complete with mon.BAR.
 168  *
 169  * - Ping a monitor prior to establishing a connection (using connect())
 170  *   and properly establish the MonClient's messenger.  This frees us
 171  *   from dealing with the complex foo that happens in connect().
 172  *
 173  * We also don't rely on MonClient as a dispatcher for this messenger,
 174  * unlike what happens with the MonClient's default messenger.  This allows
 175  * us to sandbox the whole ping, having it much as a separate entity in
 176  * the MonClient class, considerably simplifying the handling and dispatching
 177  * of messages without needing to consider monc_lock.
 178  *
 179  * Current drawback is that we will establish a messenger for each ping
 180  * we want to issue, instead of keeping a single messenger instance that
 181  * would be used for all pings.
 182  */
 183 int MonClient::ping_monitor(const string &mon_id, string *result_reply)
 184 {
 185   ldout(cct, 10) << __func__ << dendl;
 186
 187   string new_mon_id;
 188   if (monmap.contains("noname-"+mon_id)) {
 189     new_mon_id = "noname-"+mon_id;
 190   } else {
 191     new_mon_id = mon_id;
 192   }
 193
 194   if (new_mon_id.empty()) {
 195     ldout(cct, 10) << __func__ << " specified mon id is empty!" << dendl;
 196     return -EINVAL;
 197   } else if (!monmap.contains(new_mon_id)) {
 198     ldout(cct, 10) << __func__ << " no such monitor 'mon." << new_mon_id << "'"
 199                    << dendl;
 200     return -ENOENT;
 201   }
 202
 203   MonClientPinger *pinger = new MonClientPinger(cct, result_reply);
 204
 205   Messenger *smsgr = Messenger::create_client_messenger(cct, "temp_ping_client");
 206   smsgr->add_dispatcher_head(pinger);
 207   smsgr->start();
 208
 209   ConnectionRef con = smsgr->get_connection(monmap.get_inst(new_mon_id));
 210   ldout(cct, 10) << __func__ << " ping mon." << new_mon_id
 211                  << " " << con->get_peer_addr() << dendl;
 212   con->send_message(new MPing);
 213
 214   pinger->lock.Lock();
 215   int ret = pinger->wait_for_reply(cct->_conf->client_mount_timeout);
 216   if (ret == 0) {
 217     ldout(cct,10) << __func__ << " got ping reply" << dendl;
 218   } else {
 219     ret = -ret;
 220   }
 221   pinger->lock.Unlock();
 222
 223   con->mark_down();
 224   smsgr->shutdown();
 225   smsgr->wait();
 226   delete smsgr;
 227   delete pinger;
 228   return ret;
 229 }
 230
 231 bool MonClient::ms_dispatch(Message *m)
 232 {
 233   if (my_addr == entity_addr_t())
 234     my_addr = messenger->get_myaddr();
 235
 236   // we only care about these message types
 237   switch (m->get_type()) {
 238   case CEPH_MSG_MON_MAP:
 239   case CEPH_MSG_AUTH_REPLY:
 240   case CEPH_MSG_MON_SUBSCRIBE_ACK:
 241   case CEPH_MSG_MON_GET_VERSION_REPLY:
 242   case MSG_MON_COMMAND_ACK:
 243   case MSG_LOGACK:
 244     break;
 245   default:
 246     return false;
 247   }
 248
 249   Mutex::Locker lock(monc_lock);
 250
 251   if (_hunting()) {
 252     auto pending_con = pending_cons.find(m->get_source_addr());
 253     if (pending_con == pending_cons.end() ||
 254         pending_con->second.get_con() != m->get_connection()) {
 255       // ignore any messages outside hunting sessions
 256       ldout(cct, 10) << "discarding stray monitor message " << *m << dendl;
 257       m->put();
 258       return true;
 259     }
 260   } else if (!active_con || active_con->get_con() != m->get_connection()) {
 261     // ignore any messages outside our session(s)
 262     ldout(cct, 10) << "discarding stray monitor message " << *m << dendl;
 263     m->put();
 264     return true;
 265   }
 266
 267   switch (m->get_type()) {
 268   case CEPH_MSG_MON_MAP:
 269     handle_monmap(static_cast<MMonMap*>(m));
 270     if (passthrough_monmap) {
 271       return false;
 272     } else {
 273       m->put();
 274     }
 275     break;
 276   case CEPH_MSG_AUTH_REPLY:
 277     handle_auth(static_cast<MAuthReply*>(m));
 278     break;
 279   case CEPH_MSG_MON_SUBSCRIBE_ACK:
 280     handle_subscribe_ack(static_cast<MMonSubscribeAck*>(m));
 281     break;
 282   case CEPH_MSG_MON_GET_VERSION_REPLY:
 283     handle_get_version_reply(static_cast<MMonGetVersionReply*>(m));
 284     break;
 285   case MSG_MON_COMMAND_ACK:
 286     handle_mon_command_ack(static_cast<MMonCommandAck*>(m));
 287     break;
 288   case MSG_LOGACK:
 289     if (log_client) {
 290       log_client->handle_log_ack(static_cast<MLogAck*>(m));
 291       m->put();
 292       if (more_log_pending) {
 293         send_log();
 294       }
 295     } else {
 296       m->put();
 297     }
 298     break;
 299   }
 300   return true;
 301 }
 302
 303 void MonClient::send_log(bool flush)
 304 {
 305   if (log_client) {
 306     Message *lm = log_client->get_mon_log_message(flush);
 307     if (lm)
 308       _send_mon_message(lm);
 309     more_log_pending = log_client->are_pending();
 310   }
 311 }
 312
 313 void MonClient::flush_log()
 314 {
 315   Mutex::Locker l(monc_lock);
 316   send_log();
 317 }
 318
 319 /* Unlike all the other message-handling functions, we don't put away a reference
 320 * because we want to support MMonMap passthrough to other Dispatchers. */
 321 void MonClient::handle_monmap(MMonMap *m)
 322 {
 323   ldout(cct, 10) << __func__ << " " << *m << dendl;
 324   auto peer = m->get_source_addr();
 325   string cur_mon = monmap.get_name(peer);
 326
 327   bufferlist::iterator p = m->monmapbl.begin();
 328   ::decode(monmap, p);
 329
 330   ldout(cct, 10) << " got monmap " << monmap.epoch
 331                  << ", mon." << cur_mon << " is now rank " << monmap.get_rank(cur_mon)
 332                  << dendl;
 333   ldout(cct, 10) << "dump:\n";
 334   monmap.print(*_dout);
 335   *_dout << dendl;
 336
 337   _sub_got("monmap", monmap.get_epoch());
 338
 339   if (!monmap.get_addr_name(peer, cur_mon)) {
 340     ldout(cct, 10) << "mon." << cur_mon << " went away" << dendl;
 341     // can't find the mon we were talking to (above)
 342     _reopen_session();
 343   }
 344
 345   map_cond.Signal();
 346   want_monmap = false;
 347 }
 348
 349 // ----------------------
 350
 351 int MonClient::init()
 352 {
 353   ldout(cct, 10) << __func__ << dendl;
 354
 355   messenger->add_dispatcher_head(this);
 356
 357   entity_name = cct->_conf->name;
 358
 359   Mutex::Locker l(monc_lock);
 360
 361   string method;
 362   if (!cct->_conf->auth_supported.empty())
 363     method = cct->_conf->auth_supported;
 364   else if (entity_name.get_type() == CEPH_ENTITY_TYPE_OSD ||
 365            entity_name.get_type() == CEPH_ENTITY_TYPE_MDS ||
 366            entity_name.get_type() == CEPH_ENTITY_TYPE_MON)
 367     method = cct->_conf->auth_cluster_required;
 368   else
 369     method = cct->_conf->auth_client_required;
 370   auth_supported.reset(new AuthMethodList(cct, method));
 371   ldout(cct, 10) << "auth_supported " << auth_supported->get_supported_set() << " method " << method << dendl;
 372
 373   int r = 0;
 374   keyring.reset(new KeyRing); // initializing keyring anyway
 375
 376   if (auth_supported->is_supported_auth(CEPH_AUTH_CEPHX)) {
 377     r = keyring->from_ceph_context(cct);
 378     if (r == -ENOENT) {
 379       auth_supported->remove_supported_auth(CEPH_AUTH_CEPHX);
 380       if (!auth_supported->get_supported_set().empty()) {
 381         r = 0;
 382         no_keyring_disabled_cephx = true;
 383       } else {
 384         lderr(cct) << "ERROR: missing keyring, cannot use cephx for authentication" << dendl;
 385       }
 386     }
 387   }
 388
 389   if (r < 0) {
 390     return r;
 391   }
 392
 393   rotating_secrets.reset(
 394     new RotatingKeyRing(cct, cct->get_module_type(), keyring.get()));
 395
 396   initialized = true;
 397
 398   timer.init();
 399   finisher.start();
 400   schedule_tick();
 401
 402   return 0;
 403 }
 404
 405 void MonClient::shutdown()
 406 {
 407   ldout(cct, 10) << __func__ << dendl;
 408   monc_lock.Lock();
 409   while (!version_requests.empty()) {
 410     version_requests.begin()->second->context->complete(-ECANCELED);
 411     ldout(cct, 20) << __func__ << " canceling and discarding version request "
 412                    << version_requests.begin()->second << dendl;
 413     delete version_requests.begin()->second;
 414     version_requests.erase(version_requests.begin());
 415   }
 416   while (!mon_commands.empty()) {
 417     auto tid = mon_commands.begin()->first;
 418     _cancel_mon_command(tid);
 419   }
 420   while (!waiting_for_session.empty()) {
 421     ldout(cct, 20) << __func__ << " discarding pending message " << *waiting_for_session.front() << dendl;
 422     waiting_for_session.front()->put();
 423     waiting_for_session.pop_front();
 424   }
 425
 426   active_con.reset();
 427   pending_cons.clear();
 428   auth.reset();
 429
 430   monc_lock.Unlock();
 431
 432   if (initialized) {
 433     finisher.wait_for_empty();
 434     finisher.stop();
 435   }
 436   monc_lock.Lock();
 437   timer.shutdown();
 438
 439   monc_lock.Unlock();
 440 }
 441
 442 int MonClient::authenticate(double timeout)
 443 {
 444   Mutex::Locker lock(monc_lock);
 445
 446   if (active_con) {
 447     ldout(cct, 5) << "already authenticated" << dendl;
 448     return 0;
 449   }
 450
 451   _sub_want("monmap", monmap.get_epoch() ? monmap.get_epoch() + 1 : 0, 0);
 452   if (!_opened())
 453     _reopen_session();
 454
 455   utime_t until = ceph_clock_now();
 456   until += timeout;
 457   if (timeout > 0.0)
 458     ldout(cct, 10) << "authenticate will time out at " << until << dendl;
 459   while (!active_con && !authenticate_err) {
 460     if (timeout > 0.0) {
 461       int r = auth_cond.WaitUntil(monc_lock, until);
 462       if (r == ETIMEDOUT) {
 463         ldout(cct, 0) << "authenticate timed out after " << timeout << dendl;
 464         authenticate_err = -r;
 465       }
 466     } else {
 467       auth_cond.Wait(monc_lock);
 468     }
 469   }
 470
 471   if (active_con) {
 472     ldout(cct, 5) << __func__ << " success, global_id "
 473                   << active_con->get_global_id() << dendl;
 474     // active_con should not have been set if there was an error
 475     assert(authenticate_err == 0);
 476     authenticated = true;
 477   }
 478
 479   if (authenticate_err < 0 && no_keyring_disabled_cephx) {
 480     lderr(cct) << __func__ << " NOTE: no keyring found; disabled cephx authentication" << dendl;
 481   }
 482
 483   return authenticate_err;
 484 }
 485
 486 void MonClient::handle_auth(MAuthReply *m)
 487 {
 488   assert(monc_lock.is_locked());
 489   if (!_hunting()) {
 490     std::swap(active_con->get_auth(), auth);
 491     int ret = active_con->authenticate(m);
 492     m->put();
 493     std::swap(auth, active_con->get_auth());
 494     if (global_id != active_con->get_global_id()) {
 495       lderr(cct) << __func__ << " peer assigned me a different global_id: "
 496                  << active_con->get_global_id() << dendl;
 497     }
 498     if (ret != -EAGAIN) {
 499       _finish_auth(ret);
 500     }
 501     return;
 502   }
 503
 504   // hunting
 505   auto found = pending_cons.find(m->get_source_addr());
 506   assert(found != pending_cons.end());
 507   int auth_err = found->second.handle_auth(m, entity_name, want_keys,
 508                                            rotating_secrets.get());
 509   m->put();
 510   if (auth_err == -EAGAIN) {
 511     return;
 512   }
 513   if (auth_err) {
 514     pending_cons.erase(found);
 515     if (!pending_cons.empty()) {
 516       // keep trying with pending connections
 517       return;
 518     }
 519     // the last try just failed, give up.
 520   } else {
 521     auto& mc = found->second;
 522     assert(mc.have_session());
 523     active_con.reset(new MonConnection(std::move(mc)));
 524     pending_cons.clear();
 525   }
 526
 527   _finish_hunting();
 528
 529   if (!auth_err) {
 530     last_rotating_renew_sent = utime_t();
 531     while (!waiting_for_session.empty()) {
 532       _send_mon_message(waiting_for_session.front());
 533       waiting_for_session.pop_front();
 534     }
 535     _resend_mon_commands();
 536     send_log(true);
 537     if (active_con) {
 538       std::swap(auth, active_con->get_auth());
 539       global_id = active_con->get_global_id();
 540     }
 541   }
 542   _finish_auth(auth_err);
 543   if (!auth_err) {
 544     Context *cb = nullptr;
 545     if (session_established_context) {
 546       cb = session_established_context.release();
 547     }
 548     if (cb) {
 549       monc_lock.Unlock();
 550       cb->complete(0);
 551       monc_lock.Lock();
 552     }
 553   }
 554 }
 555
 556 void MonClient::_finish_auth(int auth_err)
 557 {
 558   authenticate_err = auth_err;
 559   // _resend_mon_commands() could _reopen_session() if the connected mon is not
 560   // the one the MonCommand is targeting.
 561   if (!auth_err && active_con) {
 562     assert(auth);
 563     _check_auth_tickets();
 564   }
 565   auth_cond.SignalAll();
 566 }
 567
 568 // ---------
 569
 570 void MonClient::_send_mon_message(Message *m)
 571 {
 572   assert(monc_lock.is_locked());
 573   if (active_con) {
 574     auto cur_con = active_con->get_con();
 575     ldout(cct, 10) << "_send_mon_message to mon."
 576                    << monmap.get_name(cur_con->get_peer_addr())
 577                    << " at " << cur_con->get_peer_addr() << dendl;
 578     cur_con->send_message(m);
 579   } else {
 580     waiting_for_session.push_back(m);
 581   }
 582 }
 583
 584 void MonClient::_reopen_session(int rank)
 585 {
 586   assert(monc_lock.is_locked());
 587   ldout(cct, 10) << __func__ << " rank " << rank << dendl;
 588
 589   active_con.reset();
 590   pending_cons.clear();
 591
 592   _start_hunting();
 593
 594   if (rank >= 0) {
 595     _add_conn(rank, global_id);
 596   } else {
 597     _add_conns(global_id);
 598   }
 599
 600   // throw out old queued messages
 601   while (!waiting_for_session.empty()) {
 602     waiting_for_session.front()->put();
 603     waiting_for_session.pop_front();
 604   }
 605
 606   // throw out version check requests
 607   while (!version_requests.empty()) {
 608     finisher.queue(version_requests.begin()->second->context, -EAGAIN);
 609     delete version_requests.begin()->second;
 610     version_requests.erase(version_requests.begin());
 611   }
 612
 613   for (auto& c : pending_cons) {
 614     c.second.start(monmap.get_epoch(), entity_name, *auth_supported);
 615   }
 616
 617   for (map<string,ceph_mon_subscribe_item>::iterator p = sub_sent.begin();
 618        p != sub_sent.end();
 619        ++p) {
 620     if (sub_new.count(p->first) == 0)
 621       sub_new[p->first] = p->second;
 622   }
 623   if (!sub_new.empty())
 624     _renew_subs();
 625 }
 626
 627 MonConnection& MonClient::_add_conn(unsigned rank, uint64_t global_id)
 628 {
 629   auto peer = monmap.get_addr(rank);
 630   auto conn = messenger->get_connection(monmap.get_inst(rank));
 631   MonConnection mc(cct, conn, global_id);
 632   auto inserted = pending_cons.insert(make_pair(peer, move(mc)));
 633   ldout(cct, 10) << "picked mon." << monmap.get_name(rank)
 634                  << " con " << conn
 635                  << " addr " << conn->get_peer_addr()
 636                  << dendl;
 637   return inserted.first->second;
 638 }
 639
 640 void MonClient::_add_conns(uint64_t global_id)
 641 {
 642   const unsigned num_mons = monmap.size();
 643   vector<unsigned> ranks(num_mons);
 644   for (unsigned i = 0; i < num_mons; i++) {
 645     ranks[i] = i;
 646   }
 647   std::random_device rd;
 648   std::mt19937 rng(rd());
 649   std::shuffle(ranks.begin(), ranks.end(), rng);
 650
 651   unsigned n = cct->_conf->mon_client_hunt_parallel;
 652   if (n == 0 || n > monmap.size()) {
 653      n = num_mons;
 654   }
 655   for (unsigned i = 0; i < n; i++) {
 656     _add_conn(ranks[i], global_id);
 657   }
 658 }
 659
 660 bool MonClient::ms_handle_reset(Connection *con)
 661 {
 662   Mutex::Locker lock(monc_lock);
 663
 664   if (con->get_peer_type() != CEPH_ENTITY_TYPE_MON)
 665     return false;
 666
 667   if (_hunting()) {
 668     if (pending_cons.count(con->get_peer_addr())) {
 669       ldout(cct, 10) << __func__ << " hunted mon " << con->get_peer_addr() << dendl;
 670     } else {
 671       ldout(cct, 10) << __func__ << " stray mon " << con->get_peer_addr() << dendl;
 672     }
 673     return true;
 674   } else {
 675     if (active_con && con == active_con->get_con()) {
 676       ldout(cct, 10) << __func__ << " current mon " << con->get_peer_addr() << dendl;
 677       _reopen_session();
 678       return false;
 679     } else {
 680       ldout(cct, 10) << "ms_handle_reset stray mon " << con->get_peer_addr() << dendl;
 681       return true;
 682     }
 683   }
 684 }
 685
 686 bool MonClient::_opened() const
 687 {
 688   assert(monc_lock.is_locked());
 689   return active_con || _hunting();
 690 }
 691
 692 bool MonClient::_hunting() const
 693 {
 694   return !pending_cons.empty();
 695 }
 696
 697 void MonClient::_start_hunting()
 698 {
 699   assert(!_hunting());
 700   // adjust timeouts if necessary
 701   if (!had_a_connection)
 702     return;
 703   reopen_interval_multiplier *= cct->_conf->mon_client_hunt_interval_backoff;
 704   if (reopen_interval_multiplier >
 705       cct->_conf->mon_client_hunt_interval_max_multiple) {
 706     reopen_interval_multiplier =
 707       cct->_conf->mon_client_hunt_interval_max_multiple;
 708   }
 709 }
 710
 711 void MonClient::_finish_hunting()
 712 {
 713   assert(monc_lock.is_locked());
 714   // the pending conns have been cleaned.
 715   assert(!_hunting());
 716   if (active_con) {
 717     auto con = active_con->get_con();
 718     ldout(cct, 1) << "found mon."
 719                   << monmap.get_name(con->get_peer_addr())
 720                   << dendl;
 721   } else {
 722     ldout(cct, 1) << "no mon sessions established" << dendl;
 723   }
 724
 725   had_a_connection = true;
 726   reopen_interval_multiplier /= 2.0;
 727   if (reopen_interval_multiplier < 1.0)
 728     reopen_interval_multiplier = 1.0;
 729 }
 730
 731 void MonClient::tick()
 732 {
 733   ldout(cct, 10) << __func__ << dendl;
 734
 735   _check_auth_tickets();
 736
 737   if (_hunting()) {
 738     ldout(cct, 1) << "continuing hunt" << dendl;
 739     _reopen_session();
 740   } else if (active_con) {
 741     // just renew as needed
 742     utime_t now = ceph_clock_now();
 743     auto cur_con = active_con->get_con();
 744     if (!cur_con->has_feature(CEPH_FEATURE_MON_STATEFUL_SUB)) {
 745       ldout(cct, 10) << "renew subs? (now: " << now
 746                      << "; renew after: " << sub_renew_after << ") -- "
 747                      << (now > sub_renew_after ? "yes" : "no")
 748                      << dendl;
 749       if (now > sub_renew_after)
 750         _renew_subs();
 751     }
 752
 753     cur_con->send_keepalive();
 754
 755     if (cct->_conf->mon_client_ping_timeout > 0 &&
 756         cur_con->has_feature(CEPH_FEATURE_MSGR_KEEPALIVE2)) {
 757       utime_t lk = cur_con->get_last_keepalive_ack();
 758       utime_t interval = now - lk;
 759       if (interval > cct->_conf->mon_client_ping_timeout) {
 760         ldout(cct, 1) << "no keepalive since " << lk << " (" << interval
 761                       << " seconds), reconnecting" << dendl;
 762         _reopen_session();
 763       }
 764
 765       send_log();
 766     }
 767   }
 768
 769   schedule_tick();
 770 }
 771
 772 void MonClient::schedule_tick()
 773 {
 774   struct C_Tick : public Context {
 775     MonClient *monc;
 776     explicit C_Tick(MonClient *m) : monc(m) {}
 777     void finish(int r) override {
 778       monc->tick();
 779     }
 780   };
 781
 782   if (_hunting()) {
 783     timer.add_event_after(cct->_conf->mon_client_hunt_interval
 784                           * reopen_interval_multiplier,
 785                           new C_Tick(this));
 786   } else
 787     timer.add_event_after(cct->_conf->mon_client_ping_interval, new C_Tick(this));
 788 }
 789
 790 // ---------
 791
 792 void MonClient::_renew_subs()
 793 {
 794   assert(monc_lock.is_locked());
 795   if (sub_new.empty()) {
 796     ldout(cct, 10) << __func__ << " - empty" << dendl;
 797     return;
 798   }
 799
 800   ldout(cct, 10) << __func__ << dendl;
 801   if (!_opened())
 802     _reopen_session();
 803   else {
 804     if (sub_renew_sent == utime_t())
 805       sub_renew_sent = ceph_clock_now();
 806
 807     MMonSubscribe *m = new MMonSubscribe;
 808     m->what = sub_new;
 809     _send_mon_message(m);
 810
 811     // update sub_sent with sub_new
 812     sub_new.insert(sub_sent.begin(), sub_sent.end());
 813     std::swap(sub_new, sub_sent);
 814     sub_new.clear();
 815   }
 816 }
 817
 818 void MonClient::handle_subscribe_ack(MMonSubscribeAck *m)
 819 {
 820   if (sub_renew_sent != utime_t()) {
 821     // NOTE: this is only needed for legacy (infernalis or older)
 822     // mons; see tick().
 823     sub_renew_after = sub_renew_sent;
 824     sub_renew_after += m->interval / 2.0;
 825     ldout(cct, 10) << __func__ << " sent " << sub_renew_sent << " renew after " << sub_renew_after << dendl;
 826     sub_renew_sent = utime_t();
 827   } else {
 828     ldout(cct, 10) << __func__ << " sent " << sub_renew_sent << ", ignoring" << dendl;
 829   }
 830
 831   m->put();
 832 }
 833
 834 int MonClient::_check_auth_tickets()
 835 {
 836   assert(monc_lock.is_locked());
 837   if (active_con && auth) {
 838     if (auth->need_tickets()) {
 839       ldout(cct, 10) << __func__ << " getting new tickets!" << dendl;
 840       MAuth *m = new MAuth;
 841       m->protocol = auth->get_protocol();
 842       auth->prepare_build_request();
 843       auth->build_request(m->auth_payload);
 844       _send_mon_message(m);
 845     }
 846
 847     _check_auth_rotating();
 848   }
 849   return 0;
 850 }
 851
 852 int MonClient::_check_auth_rotating()
 853 {
 854   assert(monc_lock.is_locked());
 855   if (!rotating_secrets ||
 856       !auth_principal_needs_rotating_keys(entity_name)) {
 857     ldout(cct, 20) << "_check_auth_rotating not needed by " << entity_name << dendl;
 858     return 0;
 859   }
 860
 861   if (!active_con || !auth) {
 862     ldout(cct, 10) << "_check_auth_rotating waiting for auth session" << dendl;
 863     return 0;
 864   }
 865
 866   utime_t now = ceph_clock_now();
 867   utime_t cutoff = now;
 868   cutoff -= MIN(30.0, cct->_conf->auth_service_ticket_ttl / 4.0);
 869   utime_t issued_at_lower_bound = now;
 870   issued_at_lower_bound -= cct->_conf->auth_service_ticket_ttl;
 871   if (!rotating_secrets->need_new_secrets(cutoff)) {
 872     ldout(cct, 10) << "_check_auth_rotating have uptodate secrets (they expire after " << cutoff << ")" << dendl;
 873     rotating_secrets->dump_rotating();
 874     return 0;
 875   }
 876
 877   ldout(cct, 10) << "_check_auth_rotating renewing rotating keys (they expired before " << cutoff << ")" << dendl;
 878   if (!rotating_secrets->need_new_secrets() &&
 879       rotating_secrets->need_new_secrets(issued_at_lower_bound)) {
 880     // the key has expired before it has been issued?
 881     lderr(cct) << __func__ << " possible clock skew, rotating keys expired way too early"
 882                << " (before " << issued_at_lower_bound << ")" << dendl;
 883   }
 884   if ((now > last_rotating_renew_sent) &&
 885       double(now - last_rotating_renew_sent) < 1) {
 886     ldout(cct, 10) << __func__ << " called too often (last: "
 887                    << last_rotating_renew_sent << "), skipping refresh" << dendl;
 888     return 0;
 889   }
 890   MAuth *m = new MAuth;
 891   m->protocol = auth->get_protocol();
 892   if (auth->build_rotating_request(m->auth_payload)) {
 893     last_rotating_renew_sent = now;
 894     _send_mon_message(m);
 895   } else {
 896     m->put();
 897   }
 898   return 0;
 899 }
 900
 901 int MonClient::wait_auth_rotating(double timeout)
 902 {
 903   Mutex::Locker l(monc_lock);
 904   utime_t now = ceph_clock_now();
 905   utime_t until = now;
 906   until += timeout;
 907
 908   // Must be initialized
 909   assert(auth != nullptr);
 910
 911   if (auth->get_protocol() == CEPH_AUTH_NONE)
 912     return 0;
 913
 914   if (!rotating_secrets)
 915     return 0;
 916
 917   while (auth_principal_needs_rotating_keys(entity_name) &&
 918          rotating_secrets->need_new_secrets(now)) {
 919     if (now >= until) {
 920       ldout(cct, 0) << __func__ << " timed out after " << timeout << dendl;
 921       return -ETIMEDOUT;
 922     }
 923     ldout(cct, 10) << __func__ << " waiting (until " << until << ")" << dendl;
 924     auth_cond.WaitUntil(monc_lock, until);
 925     now = ceph_clock_now();
 926   }
 927   ldout(cct, 10) << __func__ << " done" << dendl;
 928   return 0;
 929 }
 930
 931 // ---------
 932
 933 void MonClient::_send_command(MonCommand *r)
 934 {
 935   entity_addr_t peer;
 936   if (active_con) {
 937     peer = active_con->get_con()->get_peer_addr();
 938   }
 939
 940   if (r->target_rank >= 0 &&
 941       r->target_rank != monmap.get_rank(peer)) {
 942     ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd
 943                    << " wants rank " << r->target_rank
 944                    << ", reopening session"
 945                    << dendl;
 946     if (r->target_rank >= (int)monmap.size()) {
 947       ldout(cct, 10) << " target " << r->target_rank << " >= max mon " << monmap.size() << dendl;
 948       _finish_command(r, -ENOENT, "mon rank dne");
 949       return;
 950     }
 951     _reopen_session(r->target_rank);
 952     return;
 953   }
 954
 955   if (r->target_name.length() &&
 956       r->target_name != monmap.get_name(peer)) {
 957     ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd
 958                    << " wants mon " << r->target_name
 959                    << ", reopening session"
 960                    << dendl;
 961     if (!monmap.contains(r->target_name)) {
 962       ldout(cct, 10) << " target " << r->target_name << " not present in monmap" << dendl;
 963       _finish_command(r, -ENOENT, "mon dne");
 964       return;
 965     }
 966     _reopen_session(monmap.get_rank(r->target_name));
 967     return;
 968   }
 969
 970   ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
 971   MMonCommand *m = new MMonCommand(monmap.fsid);
 972   m->set_tid(r->tid);
 973   m->cmd = r->cmd;
 974   m->set_data(r->inbl);
 975   _send_mon_message(m);
 976   return;
 977 }
 978
 979 void MonClient::_resend_mon_commands()
 980 {
 981   // resend any requests
 982   for (map<uint64_t,MonCommand*>::iterator p = mon_commands.begin();
 983        p != mon_commands.end();
 984        ++p) {
 985     _send_command(p->second);
 986   }
 987 }
 988
 989 void MonClient::handle_mon_command_ack(MMonCommandAck *ack)
 990 {
 991   MonCommand *r = NULL;
 992   uint64_t tid = ack->get_tid();
 993
 994   if (tid == 0 && !mon_commands.empty()) {
 995     r = mon_commands.begin()->second;
 996     ldout(cct, 10) << __func__ << " has tid 0, assuming it is " << r->tid << dendl;
 997   } else {
 998     map<uint64_t,MonCommand*>::iterator p = mon_commands.find(tid);
 999     if (p == mon_commands.end()) {
1000       ldout(cct, 10) << __func__ << " " << ack->get_tid() << " not found" << dendl;
1001       ack->put();
1002       return;
1003     }
1004     r = p->second;
1005   }
1006
1007   ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
1008   if (r->poutbl)
1009     r->poutbl->claim(ack->get_data());
1010   _finish_command(r, ack->r, ack->rs);
1011   ack->put();
1012 }
1013
1014 int MonClient::_cancel_mon_command(uint64_t tid)
1015 {
1016   assert(monc_lock.is_locked());
1017
1018   map<ceph_tid_t, MonCommand*>::iterator it = mon_commands.find(tid);
1019   if (it == mon_commands.end()) {
1020     ldout(cct, 10) << __func__ << " tid " << tid << " dne" << dendl;
1021     return -ENOENT;
1022   }
1023
1024   ldout(cct, 10) << __func__ << " tid " << tid << dendl;
1025
1026   MonCommand *cmd = it->second;
1027   _finish_command(cmd, -ETIMEDOUT, "");
1028   return 0;
1029 }
1030
1031 void MonClient::_finish_command(MonCommand *r, int ret, string rs)
1032 {
1033   ldout(cct, 10) << __func__ << " " << r->tid << " = " << ret << " " << rs << dendl;
1034   if (r->prval)
1035     *(r->prval) = ret;
1036   if (r->prs)
1037     *(r->prs) = rs;
1038   if (r->onfinish)
1039     finisher.queue(r->onfinish, ret);
1040   mon_commands.erase(r->tid);
1041   delete r;
1042 }
1043
1044 void MonClient::start_mon_command(const vector<string>& cmd,
1045                                  const bufferlist& inbl,
1046                                  bufferlist *outbl, string *outs,
1047                                  Context *onfinish)
1048 {
1049   Mutex::Locker l(monc_lock);
1050   MonCommand *r = new MonCommand(++last_mon_command_tid);
1051   r->cmd = cmd;
1052   r->inbl = inbl;
1053   r->poutbl = outbl;
1054   r->prs = outs;
1055   r->onfinish = onfinish;
1056   if (cct->_conf->rados_mon_op_timeout > 0) {
1057     class C_CancelMonCommand : public Context
1058     {
1059       uint64_t tid;
1060       MonClient *monc;
1061       public:
1062       C_CancelMonCommand(uint64_t tid, MonClient *monc) : tid(tid), monc(monc) {}
1063       void finish(int r) override {
1064         monc->_cancel_mon_command(tid);
1065       }
1066     };
1067     r->ontimeout = new C_CancelMonCommand(r->tid, this);
1068     timer.add_event_after(cct->_conf->rados_mon_op_timeout, r->ontimeout);
1069   }
1070   mon_commands[r->tid] = r;
1071   _send_command(r);
1072 }
1073
1074 void MonClient::start_mon_command(const string &mon_name,
1075                                  const vector<string>& cmd,
1076                                  const bufferlist& inbl,
1077                                  bufferlist *outbl, string *outs,
1078                                  Context *onfinish)
1079 {
1080   Mutex::Locker l(monc_lock);
1081   MonCommand *r = new MonCommand(++last_mon_command_tid);
1082   r->target_name = mon_name;
1083   r->cmd = cmd;
1084   r->inbl = inbl;
1085   r->poutbl = outbl;
1086   r->prs = outs;
1087   r->onfinish = onfinish;
1088   mon_commands[r->tid] = r;
1089   _send_command(r);
1090 }
1091
1092 void MonClient::start_mon_command(int rank,
1093                                  const vector<string>& cmd,
1094                                  const bufferlist& inbl,
1095                                  bufferlist *outbl, string *outs,
1096                                  Context *onfinish)
1097 {
1098   Mutex::Locker l(monc_lock);
1099   MonCommand *r = new MonCommand(++last_mon_command_tid);
1100   r->target_rank = rank;
1101   r->cmd = cmd;
1102   r->inbl = inbl;
1103   r->poutbl = outbl;
1104   r->prs = outs;
1105   r->onfinish = onfinish;
1106   mon_commands[r->tid] = r;
1107   _send_command(r);
1108 }
1109
1110 // ---------
1111
1112 void MonClient::get_version(string map, version_t *newest, version_t *oldest, Context *onfinish)
1113 {
1114   version_req_d *req = new version_req_d(onfinish, newest, oldest);
1115   ldout(cct, 10) << "get_version " << map << " req " << req << dendl;
1116   Mutex::Locker l(monc_lock);
1117   MMonGetVersion *m = new MMonGetVersion();
1118   m->what = map;
1119   m->handle = ++version_req_id;
1120   version_requests[m->handle] = req;
1121   _send_mon_message(m);
1122 }
1123
1124 void MonClient::handle_get_version_reply(MMonGetVersionReply* m)
1125 {
1126   assert(monc_lock.is_locked());
1127   map<ceph_tid_t, version_req_d*>::iterator iter = version_requests.find(m->handle);
1128   if (iter == version_requests.end()) {
1129     ldout(cct, 0) << __func__ << " version request with handle " << m->handle
1130                   << " not found" << dendl;
1131   } else {
1132     version_req_d *req = iter->second;
1133     ldout(cct, 10) << __func__ << " finishing " << req << " version " << m->version << dendl;
1134     version_requests.erase(iter);
1135     if (req->newest)
1136       *req->newest = m->version;
1137     if (req->oldest)
1138       *req->oldest = m->oldest_version;
1139     finisher.queue(req->context, 0);
1140     delete req;
1141   }
1142   m->put();
1143 }
1144
1145 AuthAuthorizer* MonClient::build_authorizer(int service_id) const {
1146   Mutex::Locker l(monc_lock);
1147   if (auth) {
1148     return auth->build_authorizer(service_id);
1149   } else {
1150     ldout(cct, 0) << __func__ << " for " << ceph_entity_type_name(service_id)
1151                   << ", but no auth is available now" << dendl;
1152     return nullptr;
1153   }
1154 }
1155
1156 #define dout_subsys ceph_subsys_monc
1157 #undef dout_prefix
1158 #define dout_prefix *_dout << "monclient" << (have_session() ? ": " : "(hunting): ")
1159
1160 MonConnection::MonConnection(CephContext *cct, ConnectionRef con, uint64_t global_id)
1161   : cct(cct), con(con), global_id(global_id)
1162 {}
1163
1164 MonConnection::~MonConnection()
1165 {
1166   if (con) {
1167     con->mark_down();
1168     con.reset();
1169   }
1170 }
1171
1172 bool MonConnection::have_session() const
1173 {
1174   return state == State::HAVE_SESSION;
1175 }
1176
1177 void MonConnection::start(epoch_t epoch,
1178                          const EntityName& entity_name,
1179                          const AuthMethodList& auth_supported)
1180 {
1181   // restart authentication handshake
1182   state = State::NEGOTIATING;
1183
1184   // send an initial keepalive to ensure our timestamp is valid by the
1185   // time we are in an OPENED state (by sequencing this before
1186   // authentication).
1187   con->send_keepalive();
1188
1189   auto m = new MAuth;
1190   m->protocol = 0;
1191   m->monmap_epoch = epoch;
1192   __u8 struct_v = 1;
1193   ::encode(struct_v, m->auth_payload);
1194   ::encode(auth_supported.get_supported_set(), m->auth_payload);
1195   ::encode(entity_name, m->auth_payload);
1196   ::encode(global_id, m->auth_payload);
1197   con->send_message(m);
1198 }
1199
1200 int MonConnection::handle_auth(MAuthReply* m,
1201                                const EntityName& entity_name,
1202                                uint32_t want_keys,
1203                                RotatingKeyRing* keyring)
1204 {
1205   if (state == State::NEGOTIATING) {
1206     int r = _negotiate(m, entity_name, want_keys, keyring);
1207     if (r) {
1208       return r;
1209     }
1210     state = State::AUTHENTICATING;
1211   }
1212   int r = authenticate(m);
1213   if (!r) {
1214     state = State::HAVE_SESSION;
1215   }
1216   return r;
1217 }
1218
1219 int MonConnection::_negotiate(MAuthReply *m,
1220                               const EntityName& entity_name,
1221                               uint32_t want_keys,
1222                               RotatingKeyRing* keyring)
1223 {
1224   if (auth && (int)m->protocol == auth->get_protocol()) {
1225     // good, negotiation completed
1226     auth->reset();
1227     return 0;
1228   }
1229
1230   auth.reset(get_auth_client_handler(cct, m->protocol, keyring));
1231   if (!auth) {
1232     ldout(cct, 10) << "no handler for protocol " << m->protocol << dendl;
1233     if (m->result == -ENOTSUP) {
1234       ldout(cct, 10) << "none of our auth protocols are supported by the server"
1235                      << dendl;
1236     }
1237     return m->result;
1238   }
1239
1240   // do not request MGR key unless the mon has the SERVER_KRAKEN
1241   // feature.  otherwise it will give us an auth error.  note that
1242   // we have to use the FEATUREMASK because pre-jewel the kraken
1243   // feature bit was used for something else.
1244   if ((want_keys & CEPH_ENTITY_TYPE_MGR) &&
1245       !(m->get_connection()->has_features(CEPH_FEATUREMASK_SERVER_KRAKEN))) {
1246     ldout(cct, 1) << __func__
1247                   << " not requesting MGR keys from pre-kraken monitor"
1248                   << dendl;
1249     want_keys &= ~CEPH_ENTITY_TYPE_MGR;
1250   }
1251   auth->set_want_keys(want_keys);
1252   auth->init(entity_name);
1253   auth->set_global_id(global_id);
1254   return 0;
1255 }
1256
1257 int MonConnection::authenticate(MAuthReply *m)
1258 {
1259   assert(auth);
1260   if (!m->global_id) {
1261     ldout(cct, 1) << "peer sent an invalid global_id" << dendl;
1262   }
1263   if (m->global_id != global_id) {
1264     // it's a new session
1265     auth->reset();
1266     global_id = m->global_id;
1267     auth->set_global_id(global_id);
1268     ldout(cct, 10) << "my global_id is " << m->global_id << dendl;
1269   }
1270   auto p = m->result_bl.begin();
1271   int ret = auth->handle_response(m->result, p);
1272   if (ret == -EAGAIN) {
1273     auto ma = new MAuth;
1274     ma->protocol = auth->get_protocol();
1275     auth->prepare_build_request();
1276     auth->build_request(ma->auth_payload);
1277     con->send_message(ma);
1278   }
1279   return ret;
1280 }