ceph/src/mon/MonClient.h

   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 #ifndef CEPH_MONCLIENT_H
  15 #define CEPH_MONCLIENT_H
  16
  17 #include <memory>
  18
  19 #include "msg/Messenger.h"
  20
  21 #include "MonMap.h"
  22 #include "MonSub.h"
  23
  24 #include "common/Timer.h"
  25 #include "common/Finisher.h"
  26 #include "common/config.h"
  27
  28 #include "auth/AuthClient.h"
  29 #include "auth/AuthServer.h"
  30
  31 class MMonMap;
  32 class MConfig;
  33 class MMonGetVersionReply;
  34 struct MMonSubscribeAck;
  35 class MMonCommandAck;
  36 struct MAuthReply;
  37 class LogClient;
  38 class AuthAuthorizer;
  39 class AuthClientHandler;
  40 class AuthMethodList;
  41 class AuthRegistry;
  42 class KeyRing;
  43 class RotatingKeyRing;
  44
  45 class MonConnection {
  46 public:
  47   MonConnection(CephContext *cct,
  48                 ConnectionRef conn,
  49                 uint64_t global_id,
  50                 AuthRegistry *auth_registry);
  51   ~MonConnection();
  52   MonConnection(MonConnection&& rhs) = default;
  53   MonConnection& operator=(MonConnection&&) = default;
  54   MonConnection(const MonConnection& rhs) = delete;
  55   MonConnection& operator=(const MonConnection&) = delete;
  56   int handle_auth(MAuthReply *m,
  57                   const EntityName& entity_name,
  58                   uint32_t want_keys,
  59                   RotatingKeyRing* keyring);
  60   int authenticate(MAuthReply *m);
  61   void start(epoch_t epoch,
  62              const EntityName& entity_name);
  63   bool have_session() const;
  64   uint64_t get_global_id() const {
  65     return global_id;
  66   }
  67   ConnectionRef get_con() {
  68     return con;
  69   }
  70   std::unique_ptr<AuthClientHandler>& get_auth() {
  71     return auth;
  72   }
  73
  74   int get_auth_request(
  75     uint32_t *method,
  76     std::vector<uint32_t> *preferred_modes,
  77     bufferlist *out,
  78     const EntityName& entity_name,
  79     uint32_t want_keys,
  80     RotatingKeyRing* keyring);
  81   int handle_auth_reply_more(
  82     AuthConnectionMeta *auth_meta,
  83     const bufferlist& bl,
  84     bufferlist *reply);
  85   int handle_auth_done(
  86     AuthConnectionMeta *auth_meta,
  87     uint64_t global_id,
  88     const bufferlist& bl,
  89     CryptoKey *session_key,
  90     std::string *connection_secret);
  91   int handle_auth_bad_method(
  92     uint32_t old_auth_method,
  93     int result,
  94     const std::vector<uint32_t>& allowed_methods,
  95     const std::vector<uint32_t>& allowed_modes);
  96
  97   bool is_con(Connection *c) const {
  98     return con.get() == c;
  99   }
 100
 101 private:
 102   int _negotiate(MAuthReply *m,
 103                  const EntityName& entity_name,
 104                  uint32_t want_keys,
 105                  RotatingKeyRing* keyring);
 106   int _init_auth(uint32_t method,
 107                  const EntityName& entity_name,
 108                  uint32_t want_keys,
 109                  RotatingKeyRing* keyring,
 110                  bool msgr2);
 111
 112 private:
 113   CephContext *cct;
 114   enum class State {
 115     NONE,
 116     NEGOTIATING,       // v1 only
 117     AUTHENTICATING,    // v1 and v2
 118     HAVE_SESSION,
 119   };
 120   State state = State::NONE;
 121   ConnectionRef con;
 122   int auth_method = -1;
 123   utime_t auth_start;
 124
 125   std::unique_ptr<AuthClientHandler> auth;
 126   uint64_t global_id;
 127
 128   AuthRegistry *auth_registry;
 129 };
 130
 131
 132 struct MonClientPinger : public Dispatcher,
 133                          public AuthClient {
 134
 135   Mutex lock;
 136   Cond ping_recvd_cond;
 137   string *result;
 138   bool done;
 139   RotatingKeyRing *keyring;
 140   std::unique_ptr<MonConnection> mc;
 141
 142   MonClientPinger(CephContext *cct_,
 143                   RotatingKeyRing *keyring,
 144                   string *res_) :
 145     Dispatcher(cct_),
 146     lock("MonClientPinger::lock"),
 147     result(res_),
 148     done(false),
 149     keyring(keyring)
 150   { }
 151
 152   int wait_for_reply(double timeout = 0.0) {
 153     utime_t until = ceph_clock_now();
 154     until += (timeout > 0 ? timeout : cct->_conf->client_mount_timeout);
 155     done = false;
 156
 157     int ret = 0;
 158     while (!done) {
 159       ret = ping_recvd_cond.WaitUntil(lock, until);
 160       if (ret == ETIMEDOUT)
 161         break;
 162     }
 163     return ret;
 164   }
 165
 166   bool ms_dispatch(Message *m) override {
 167     std::lock_guard l(lock);
 168     if (m->get_type() != CEPH_MSG_PING)
 169       return false;
 170
 171     bufferlist &payload = m->get_payload();
 172     if (result && payload.length() > 0) {
 173       auto p = std::cbegin(payload);
 174       decode(*result, p);
 175     }
 176     done = true;
 177     ping_recvd_cond.SignalAll();
 178     m->put();
 179     return true;
 180   }
 181   bool ms_handle_reset(Connection *con) override {
 182     std::lock_guard l(lock);
 183     done = true;
 184     ping_recvd_cond.SignalAll();
 185     return true;
 186   }
 187   void ms_handle_remote_reset(Connection *con) override {}
 188   bool ms_handle_refused(Connection *con) override {
 189     return false;
 190   }
 191
 192   // AuthClient
 193   int get_auth_request(
 194     Connection *con,
 195     AuthConnectionMeta *auth_meta,
 196     uint32_t *auth_method,
 197     std::vector<uint32_t> *preferred_modes,
 198     bufferlist *bl) override {
 199     return mc->get_auth_request(auth_method, preferred_modes, bl,
 200                                 cct->_conf->name, 0, keyring);
 201   }
 202   int handle_auth_reply_more(
 203     Connection *con,
 204     AuthConnectionMeta *auth_meta,
 205     const bufferlist& bl,
 206     bufferlist *reply) override {
 207     return mc->handle_auth_reply_more(auth_meta, bl, reply);
 208   }
 209   int handle_auth_done(
 210     Connection *con,
 211     AuthConnectionMeta *auth_meta,
 212     uint64_t global_id,
 213     uint32_t con_mode,
 214     const bufferlist& bl,
 215     CryptoKey *session_key,
 216     std::string *connection_secret) override {
 217     return mc->handle_auth_done(auth_meta, global_id, bl,
 218                                 session_key, connection_secret);
 219   }
 220   int handle_auth_bad_method(
 221     Connection *con,
 222     AuthConnectionMeta *auth_meta,
 223     uint32_t old_auth_method,
 224     int result,
 225     const std::vector<uint32_t>& allowed_methods,
 226     const std::vector<uint32_t>& allowed_modes) override {
 227     return mc->handle_auth_bad_method(old_auth_method, result,
 228                                       allowed_methods, allowed_modes);
 229   }
 230 };
 231
 232
 233 class MonClient : public Dispatcher,
 234                   public AuthClient,
 235                   public AuthServer /* for mgr, osd, mds */ {
 236 public:
 237   MonMap monmap;
 238   map<string,string> config_mgr;
 239
 240 private:
 241   Messenger *messenger;
 242
 243   std::unique_ptr<MonConnection> active_con;
 244   std::map<entity_addrvec_t, MonConnection> pending_cons;
 245
 246   EntityName entity_name;
 247
 248   mutable Mutex monc_lock;
 249   SafeTimer timer;
 250   Finisher finisher;
 251
 252   bool initialized;
 253   bool stopping = false;
 254
 255   LogClient *log_client;
 256   bool more_log_pending;
 257
 258   void send_log(bool flush = false);
 259
 260   bool ms_dispatch(Message *m) override;
 261   bool ms_handle_reset(Connection *con) override;
 262   void ms_handle_remote_reset(Connection *con) override {}
 263   bool ms_handle_refused(Connection *con) override { return false; }
 264
 265   void handle_monmap(MMonMap *m);
 266   void handle_config(MConfig *m);
 267
 268   void handle_auth(MAuthReply *m);
 269
 270   // monitor session
 271   void tick();
 272   void schedule_tick();
 273
 274   // monclient
 275   bool want_monmap;
 276   Cond map_cond;
 277   bool passthrough_monmap = false;
 278   bool got_config = false;
 279
 280   // authenticate
 281   std::unique_ptr<AuthClientHandler> auth;
 282   uint32_t want_keys = 0;
 283   uint64_t global_id = 0;
 284   Cond auth_cond;
 285   int authenticate_err = 0;
 286   bool authenticated = false;
 287
 288   list<Message*> waiting_for_session;
 289   utime_t last_rotating_renew_sent;
 290   std::unique_ptr<Context> session_established_context;
 291   bool had_a_connection;
 292   double reopen_interval_multiplier;
 293
 294   Dispatcher *handle_authentication_dispatcher = nullptr;
 295
 296   bool _opened() const;
 297   bool _hunting() const;
 298   void _start_hunting();
 299   void _finish_hunting(int auth_err);
 300   void _finish_auth(int auth_err);
 301   void _reopen_session(int rank = -1);
 302   MonConnection& _add_conn(unsigned rank, uint64_t global_id);
 303   void _un_backoff();
 304   void _add_conns(uint64_t global_id);
 305   void _send_mon_message(Message *m);
 306
 307   std::map<entity_addrvec_t, MonConnection>::iterator _find_pending_con(
 308     const ConnectionRef& con) {
 309     for (auto i = pending_cons.begin(); i != pending_cons.end(); ++i) {
 310       if (i->second.get_con() == con) {
 311         return i;
 312       }
 313     }
 314     return pending_cons.end();
 315   }
 316
 317 public:
 318   // AuthClient
 319   int get_auth_request(
 320     Connection *con,
 321     AuthConnectionMeta *auth_meta,
 322     uint32_t *method,
 323     std::vector<uint32_t> *preferred_modes,
 324     bufferlist *bl) override;
 325   int handle_auth_reply_more(
 326     Connection *con,
 327     AuthConnectionMeta *auth_meta,
 328     const bufferlist& bl,
 329     bufferlist *reply) override;
 330   int handle_auth_done(
 331     Connection *con,
 332     AuthConnectionMeta *auth_meta,
 333     uint64_t global_id,
 334     uint32_t con_mode,
 335     const bufferlist& bl,
 336     CryptoKey *session_key,
 337     std::string *connection_secret) override;
 338   int handle_auth_bad_method(
 339     Connection *con,
 340     AuthConnectionMeta *auth_meta,
 341     uint32_t old_auth_method,
 342     int result,
 343     const std::vector<uint32_t>& allowed_methods,
 344     const std::vector<uint32_t>& allowed_modes) override;
 345   // AuthServer
 346   int handle_auth_request(
 347     Connection *con,
 348     AuthConnectionMeta *auth_meta,
 349     bool more,
 350     uint32_t auth_method,
 351     const bufferlist& bl,
 352     bufferlist *reply);
 353
 354   void set_entity_name(EntityName name) { entity_name = name; }
 355   void set_handle_authentication_dispatcher(Dispatcher *d) {
 356     handle_authentication_dispatcher = d;
 357   }
 358   int _check_auth_tickets();
 359   int _check_auth_rotating();
 360   int wait_auth_rotating(double timeout);
 361
 362   int authenticate(double timeout=0.0);
 363   bool is_authenticated() const {return authenticated;}
 364
 365   bool is_connected() const { return active_con != nullptr; }
 366
 367   /**
 368    * Try to flush as many log messages as we can in a single
 369    * message.  Use this before shutting down to transmit your
 370    * last message.
 371    */
 372   void flush_log();
 373
 374 private:
 375   // mon subscriptions
 376   MonSub sub;
 377   void _renew_subs();
 378   void handle_subscribe_ack(MMonSubscribeAck* m);
 379
 380 public:
 381   void renew_subs() {
 382     std::lock_guard l(monc_lock);
 383     _renew_subs();
 384   }
 385   bool sub_want(string what, version_t start, unsigned flags) {
 386     std::lock_guard l(monc_lock);
 387     return sub.want(what, start, flags);
 388   }
 389   void sub_got(string what, version_t have) {
 390     std::lock_guard l(monc_lock);
 391     sub.got(what, have);
 392   }
 393   void sub_unwant(string what) {
 394     std::lock_guard l(monc_lock);
 395     sub.unwant(what);
 396   }
 397   bool sub_want_increment(string what, version_t start, unsigned flags) {
 398     std::lock_guard l(monc_lock);
 399     return sub.inc_want(what, start, flags);
 400   }
 401
 402   std::unique_ptr<KeyRing> keyring;
 403   std::unique_ptr<RotatingKeyRing> rotating_secrets;
 404
 405  public:
 406   explicit MonClient(CephContext *cct_);
 407   MonClient(const MonClient &) = delete;
 408   MonClient& operator=(const MonClient &) = delete;
 409   ~MonClient() override;
 410
 411   int init();
 412   void shutdown();
 413
 414   void set_log_client(LogClient *clog) {
 415     log_client = clog;
 416   }
 417
 418   int build_initial_monmap();
 419   int get_monmap();
 420   int get_monmap_and_config();
 421   /**
 422    * If you want to see MonMap messages, set this and
 423    * the MonClient will tell the Messenger it hasn't
 424    * dealt with it.
 425    * Note that if you do this, *you* are of course responsible for
 426    * putting the message reference!
 427    */
 428   void set_passthrough_monmap() {
 429     std::lock_guard l(monc_lock);
 430     passthrough_monmap = true;
 431   }
 432   void unset_passthrough_monmap() {
 433     std::lock_guard l(monc_lock);
 434     passthrough_monmap = false;
 435   }
 436   /**
 437    * Ping monitor with ID @p mon_id and record the resulting
 438    * reply in @p result_reply.
 439    *
 440    * @param[in]  mon_id Target monitor's ID
 441    * @param[out] result_reply reply from mon.ID, if param != NULL
 442    * @returns    0 in case of success; < 0 in case of error,
 443    *             -ETIMEDOUT if monitor didn't reply before timeout
 444    *             expired (default: conf->client_mount_timeout).
 445    */
 446   int ping_monitor(const string &mon_id, string *result_reply);
 447
 448   void send_mon_message(Message *m) {
 449     std::lock_guard l(monc_lock);
 450     _send_mon_message(m);
 451   }
 452   /**
 453    * If you specify a callback, you should not call
 454    * reopen_session() again until it has been triggered. The MonClient
 455    * will behave, but the first callback could be triggered after
 456    * the session has been killed and the MonClient has started trying
 457    * to reconnect to another monitor.
 458    */
 459   void reopen_session(Context *cb=NULL) {
 460     std::lock_guard l(monc_lock);
 461     if (cb) {
 462       session_established_context.reset(cb);
 463     }
 464     _reopen_session();
 465   }
 466
 467   const uuid_d& get_fsid() const {
 468     return monmap.fsid;
 469   }
 470
 471   entity_addrvec_t get_mon_addrs(unsigned i) const {
 472     std::lock_guard l(monc_lock);
 473     if (i < monmap.size())
 474       return monmap.get_addrs(i);
 475     return entity_addrvec_t();
 476   }
 477   int get_num_mon() const {
 478     std::lock_guard l(monc_lock);
 479     return monmap.size();
 480   }
 481
 482   uint64_t get_global_id() const {
 483     std::lock_guard l(monc_lock);
 484     return global_id;
 485   }
 486
 487   void set_messenger(Messenger *m) { messenger = m; }
 488   entity_addrvec_t get_myaddrs() const { return messenger->get_myaddrs(); }
 489   AuthAuthorizer* build_authorizer(int service_id) const;
 490
 491   void set_want_keys(uint32_t want) {
 492     want_keys = want;
 493   }
 494
 495   // admin commands
 496 private:
 497   uint64_t last_mon_command_tid;
 498
 499   struct MonCommand {
 500     string target_name;
 501     int target_rank;
 502     uint64_t tid;
 503     vector<string> cmd;
 504     bufferlist inbl;
 505     bufferlist *poutbl;
 506     string *prs;
 507     int *prval;
 508     Context *onfinish, *ontimeout;
 509
 510     explicit MonCommand(uint64_t t)
 511       : target_rank(-1),
 512         tid(t),
 513         poutbl(NULL), prs(NULL), prval(NULL), onfinish(NULL), ontimeout(NULL)
 514     {}
 515   };
 516   map<uint64_t,MonCommand*> mon_commands;
 517
 518   void _send_command(MonCommand *r);
 519   void _resend_mon_commands();
 520   int _cancel_mon_command(uint64_t tid);
 521   void _finish_command(MonCommand *r, int ret, string rs);
 522   void _finish_auth();
 523   void handle_mon_command_ack(MMonCommandAck *ack);
 524
 525 public:
 526   void start_mon_command(const vector<string>& cmd, const bufferlist& inbl,
 527                         bufferlist *outbl, string *outs,
 528                         Context *onfinish);
 529   void start_mon_command(int mon_rank,
 530                         const vector<string>& cmd, const bufferlist& inbl,
 531                         bufferlist *outbl, string *outs,
 532                         Context *onfinish);
 533   void start_mon_command(const string &mon_name,  ///< mon name, with mon. prefix
 534                         const vector<string>& cmd, const bufferlist& inbl,
 535                         bufferlist *outbl, string *outs,
 536                         Context *onfinish);
 537
 538   // version requests
 539 public:
 540   /**
 541    * get latest known version(s) of cluster map
 542    *
 543    * @param map string name of map (e.g., 'osdmap')
 544    * @param newest pointer where newest map version will be stored
 545    * @param oldest pointer where oldest map version will be stored
 546    * @param onfinish context that will be triggered on completion
 547    * @return (via context) 0 on success, -EAGAIN if we need to resubmit our request
 548    */
 549   void get_version(string map, version_t *newest, version_t *oldest, Context *onfinish);
 550   /**
 551    * Run a callback within our lock, with a reference
 552    * to the MonMap
 553    */
 554   template<typename Callback, typename...Args>
 555   auto with_monmap(Callback&& cb, Args&&...args) const ->
 556     decltype(cb(monmap, std::forward<Args>(args)...)) {
 557     std::lock_guard l(monc_lock);
 558     return std::forward<Callback>(cb)(monmap, std::forward<Args>(args)...);
 559   }
 560
 561   void register_config_callback(md_config_t::config_callback fn);
 562   void register_config_notify_callback(std::function<void(void)> f) {
 563     config_notify_cb = f;
 564   }
 565   md_config_t::config_callback get_config_callback();
 566
 567 private:
 568   struct version_req_d {
 569     Context *context;
 570     version_t *newest, *oldest;
 571     version_req_d(Context *con, version_t *n, version_t *o) : context(con),newest(n), oldest(o) {}
 572   };
 573
 574   map<ceph_tid_t, version_req_d*> version_requests;
 575   ceph_tid_t version_req_id;
 576   void handle_get_version_reply(MMonGetVersionReply* m);
 577
 578   md_config_t::config_callback config_cb;
 579   std::function<void(void)> config_notify_cb;
 580 };
 581
 582 #endif