diff options
Diffstat (limited to '')
| -rw-r--r-- | src/mon/MonClient.h | 583 |
1 files changed, 583 insertions, 0 deletions
diff --git a/src/mon/MonClient.h b/src/mon/MonClient.h new file mode 100644 index 00000000..abdcd315 --- /dev/null +++ b/src/mon/MonClient.h @@ -0,0 +1,583 @@ +// -*- 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. + * + */ +#ifndef CEPH_MONCLIENT_H +#define CEPH_MONCLIENT_H + +#include <memory> + +#include "msg/Messenger.h" + +#include "MonMap.h" +#include "MonSub.h" + +#include "common/Timer.h" +#include "common/Finisher.h" +#include "common/config.h" + +#include "auth/AuthClient.h" +#include "auth/AuthServer.h" + +class MMonMap; +class MConfig; +class MMonGetVersionReply; +struct MMonSubscribeAck; +class MMonCommandAck; +struct MAuthReply; +class LogClient; +class AuthAuthorizer; +class AuthClientHandler; +class AuthMethodList; +class AuthRegistry; +class KeyRing; +class RotatingKeyRing; + +class MonConnection { +public: + MonConnection(CephContext *cct, + ConnectionRef conn, + uint64_t global_id, + AuthRegistry *auth_registry); + ~MonConnection(); + MonConnection(MonConnection&& rhs) = default; + MonConnection& operator=(MonConnection&&) = default; + MonConnection(const MonConnection& rhs) = delete; + MonConnection& operator=(const MonConnection&) = delete; + int handle_auth(MAuthReply *m, + const EntityName& entity_name, + uint32_t want_keys, + RotatingKeyRing* keyring); + int authenticate(MAuthReply *m); + void start(epoch_t epoch, + const EntityName& entity_name); + bool have_session() const; + uint64_t get_global_id() const { + return global_id; + } + ConnectionRef get_con() { + return con; + } + std::unique_ptr<AuthClientHandler>& get_auth() { + return auth; + } + + int get_auth_request( + uint32_t *method, + std::vector<uint32_t> *preferred_modes, + bufferlist *out, + const EntityName& entity_name, + uint32_t want_keys, + RotatingKeyRing* keyring); + int handle_auth_reply_more( + AuthConnectionMeta *auth_meta, + const bufferlist& bl, + bufferlist *reply); + int handle_auth_done( + AuthConnectionMeta *auth_meta, + uint64_t global_id, + const bufferlist& bl, + CryptoKey *session_key, + std::string *connection_secret); + int 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); + + bool is_con(Connection *c) const { + return con.get() == c; + } + +private: + int _negotiate(MAuthReply *m, + const EntityName& entity_name, + uint32_t want_keys, + RotatingKeyRing* keyring); + int _init_auth(uint32_t method, + const EntityName& entity_name, + uint32_t want_keys, + RotatingKeyRing* keyring, + bool msgr2); + +private: + CephContext *cct; + enum class State { + NONE, + NEGOTIATING, // v1 only + AUTHENTICATING, // v1 and v2 + HAVE_SESSION, + }; + State state = State::NONE; + ConnectionRef con; + int auth_method = -1; + utime_t auth_start; + + std::unique_ptr<AuthClientHandler> auth; + uint64_t global_id; + + AuthRegistry *auth_registry; +}; + + +struct MonClientPinger : public Dispatcher, + public AuthClient { + + Mutex lock; + Cond ping_recvd_cond; + string *result; + bool done; + RotatingKeyRing *keyring; + std::unique_ptr<MonConnection> mc; + + MonClientPinger(CephContext *cct_, + RotatingKeyRing *keyring, + string *res_) : + Dispatcher(cct_), + lock("MonClientPinger::lock"), + result(res_), + done(false), + keyring(keyring) + { } + + int wait_for_reply(double timeout = 0.0) { + utime_t until = ceph_clock_now(); + until += (timeout > 0 ? timeout : cct->_conf->client_mount_timeout); + done = false; + + int ret = 0; + while (!done) { + ret = ping_recvd_cond.WaitUntil(lock, until); + if (ret == ETIMEDOUT) + break; + } + return ret; + } + + bool ms_dispatch(Message *m) override { + std::lock_guard l(lock); + if (m->get_type() != CEPH_MSG_PING) + return false; + + bufferlist &payload = m->get_payload(); + if (result && payload.length() > 0) { + auto p = std::cbegin(payload); + decode(*result, p); + } + done = true; + ping_recvd_cond.SignalAll(); + m->put(); + return true; + } + bool ms_handle_reset(Connection *con) override { + std::lock_guard l(lock); + done = true; + ping_recvd_cond.SignalAll(); + return true; + } + void ms_handle_remote_reset(Connection *con) override {} + bool ms_handle_refused(Connection *con) override { + return false; + } + + // AuthClient + int get_auth_request( + Connection *con, + AuthConnectionMeta *auth_meta, + uint32_t *auth_method, + std::vector<uint32_t> *preferred_modes, + bufferlist *bl) override { + return mc->get_auth_request(auth_method, preferred_modes, bl, + cct->_conf->name, 0, keyring); + } + int handle_auth_reply_more( + Connection *con, + AuthConnectionMeta *auth_meta, + const bufferlist& bl, + bufferlist *reply) override { + return mc->handle_auth_reply_more(auth_meta, bl, reply); + } + int 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) override { + return mc->handle_auth_done(auth_meta, global_id, bl, + session_key, connection_secret); + } + int 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) override { + return mc->handle_auth_bad_method(old_auth_method, result, + allowed_methods, allowed_modes); + } +}; + + +class MonClient : public Dispatcher, + public AuthClient, + public AuthServer /* for mgr, osd, mds */ { +public: + MonMap monmap; + map<string,string> config_mgr; + +private: + Messenger *messenger; + + std::unique_ptr<MonConnection> active_con; + std::map<entity_addrvec_t, MonConnection> pending_cons; + + EntityName entity_name; + + mutable Mutex monc_lock; + SafeTimer timer; + Finisher finisher; + + bool initialized; + bool stopping = false; + + LogClient *log_client; + bool more_log_pending; + + void send_log(bool flush = false); + + bool ms_dispatch(Message *m) override; + bool ms_handle_reset(Connection *con) override; + void ms_handle_remote_reset(Connection *con) override {} + bool ms_handle_refused(Connection *con) override { return false; } + + void handle_monmap(MMonMap *m); + void handle_config(MConfig *m); + + void handle_auth(MAuthReply *m); + + // monitor session + void tick(); + void schedule_tick(); + + // monclient + bool want_monmap; + Cond map_cond; + bool passthrough_monmap = false; + bool got_config = false; + + // authenticate + std::unique_ptr<AuthClientHandler> auth; + uint32_t want_keys = 0; + uint64_t global_id = 0; + Cond auth_cond; + int authenticate_err = 0; + bool authenticated = false; + + list<Message*> waiting_for_session; + utime_t last_rotating_renew_sent; + std::unique_ptr<Context> session_established_context; + bool had_a_connection; + double reopen_interval_multiplier; + + Dispatcher *handle_authentication_dispatcher = nullptr; + + bool _opened() const; + bool _hunting() const; + void _start_hunting(); + void _finish_hunting(int auth_err); + void _finish_auth(int auth_err); + void _reopen_session(int rank = -1); + MonConnection& _add_conn(unsigned rank); + void _add_conns(); + void _un_backoff(); + void _send_mon_message(Message *m); + + std::map<entity_addrvec_t, MonConnection>::iterator _find_pending_con( + const ConnectionRef& con) { + for (auto i = pending_cons.begin(); i != pending_cons.end(); ++i) { + if (i->second.get_con() == con) { + return i; + } + } + return pending_cons.end(); + } + +public: + // AuthClient + int get_auth_request( + Connection *con, + AuthConnectionMeta *auth_meta, + uint32_t *method, + std::vector<uint32_t> *preferred_modes, + bufferlist *bl) override; + int handle_auth_reply_more( + Connection *con, + AuthConnectionMeta *auth_meta, + const bufferlist& bl, + bufferlist *reply) override; + int 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) override; + int 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) override; + // AuthServer + int handle_auth_request( + Connection *con, + AuthConnectionMeta *auth_meta, + bool more, + uint32_t auth_method, + const bufferlist& bl, + bufferlist *reply); + + void set_entity_name(EntityName name) { entity_name = name; } + void set_handle_authentication_dispatcher(Dispatcher *d) { + handle_authentication_dispatcher = d; + } + int _check_auth_tickets(); + int _check_auth_rotating(); + int wait_auth_rotating(double timeout); + + int authenticate(double timeout=0.0); + bool is_authenticated() const {return authenticated;} + + bool is_connected() const { return active_con != nullptr; } + + /** + * Try to flush as many log messages as we can in a single + * message. Use this before shutting down to transmit your + * last message. + */ + void flush_log(); + +private: + // mon subscriptions + MonSub sub; + void _renew_subs(); + void handle_subscribe_ack(MMonSubscribeAck* m); + +public: + void renew_subs() { + std::lock_guard l(monc_lock); + _renew_subs(); + } + bool sub_want(string what, version_t start, unsigned flags) { + std::lock_guard l(monc_lock); + return sub.want(what, start, flags); + } + void sub_got(string what, version_t have) { + std::lock_guard l(monc_lock); + sub.got(what, have); + } + void sub_unwant(string what) { + std::lock_guard l(monc_lock); + sub.unwant(what); + } + bool sub_want_increment(string what, version_t start, unsigned flags) { + std::lock_guard l(monc_lock); + return sub.inc_want(what, start, flags); + } + + std::unique_ptr<KeyRing> keyring; + std::unique_ptr<RotatingKeyRing> rotating_secrets; + + public: + explicit MonClient(CephContext *cct_); + MonClient(const MonClient &) = delete; + MonClient& operator=(const MonClient &) = delete; + ~MonClient() override; + + int init(); + void shutdown(); + + void set_log_client(LogClient *clog) { + log_client = clog; + } + + int build_initial_monmap(); + int get_monmap(); + int get_monmap_and_config(); + /** + * If you want to see MonMap messages, set this and + * the MonClient will tell the Messenger it hasn't + * dealt with it. + * Note that if you do this, *you* are of course responsible for + * putting the message reference! + */ + void set_passthrough_monmap() { + std::lock_guard l(monc_lock); + passthrough_monmap = true; + } + void unset_passthrough_monmap() { + std::lock_guard l(monc_lock); + passthrough_monmap = false; + } + /** + * Ping monitor with ID @p mon_id and record the resulting + * reply in @p result_reply. + * + * @param[in] mon_id Target monitor's ID + * @param[out] result_reply reply from mon.ID, if param != NULL + * @returns 0 in case of success; < 0 in case of error, + * -ETIMEDOUT if monitor didn't reply before timeout + * expired (default: conf->client_mount_timeout). + */ + int ping_monitor(const string &mon_id, string *result_reply); + + void send_mon_message(Message *m) { + std::lock_guard l(monc_lock); + _send_mon_message(m); + } + /** + * If you specify a callback, you should not call + * reopen_session() again until it has been triggered. The MonClient + * will behave, but the first callback could be triggered after + * the session has been killed and the MonClient has started trying + * to reconnect to another monitor. + */ + void reopen_session(Context *cb=NULL) { + std::lock_guard l(monc_lock); + if (cb) { + session_established_context.reset(cb); + } + _reopen_session(); + } + + const uuid_d& get_fsid() const { + return monmap.fsid; + } + + entity_addrvec_t get_mon_addrs(unsigned i) const { + std::lock_guard l(monc_lock); + if (i < monmap.size()) + return monmap.get_addrs(i); + return entity_addrvec_t(); + } + int get_num_mon() const { + std::lock_guard l(monc_lock); + return monmap.size(); + } + + uint64_t get_global_id() const { + std::lock_guard l(monc_lock); + return global_id; + } + + void set_messenger(Messenger *m) { messenger = m; } + entity_addrvec_t get_myaddrs() const { return messenger->get_myaddrs(); } + AuthAuthorizer* build_authorizer(int service_id) const; + + void set_want_keys(uint32_t want) { + want_keys = want; + } + + // admin commands +private: + uint64_t last_mon_command_tid; + + struct MonCommand { + string target_name; + int target_rank; + unsigned send_attempts = 0; + uint64_t tid; + vector<string> cmd; + bufferlist inbl; + bufferlist *poutbl; + string *prs; + int *prval; + Context *onfinish, *ontimeout; + + explicit MonCommand(uint64_t t) + : target_rank(-1), + tid(t), + poutbl(NULL), prs(NULL), prval(NULL), onfinish(NULL), ontimeout(NULL) + {} + }; + map<uint64_t,MonCommand*> mon_commands; + + void _send_command(MonCommand *r); + void _resend_mon_commands(); + int _cancel_mon_command(uint64_t tid); + void _finish_command(MonCommand *r, int ret, string rs); + void _finish_auth(); + void handle_mon_command_ack(MMonCommandAck *ack); + +public: + void start_mon_command(const vector<string>& cmd, const bufferlist& inbl, + bufferlist *outbl, string *outs, + Context *onfinish); + void start_mon_command(int mon_rank, + const vector<string>& cmd, const bufferlist& inbl, + bufferlist *outbl, string *outs, + Context *onfinish); + void start_mon_command(const string &mon_name, ///< mon name, with mon. prefix + const vector<string>& cmd, const bufferlist& inbl, + bufferlist *outbl, string *outs, + Context *onfinish); + + // version requests +public: + /** + * get latest known version(s) of cluster map + * + * @param map string name of map (e.g., 'osdmap') + * @param newest pointer where newest map version will be stored + * @param oldest pointer where oldest map version will be stored + * @param onfinish context that will be triggered on completion + * @return (via context) 0 on success, -EAGAIN if we need to resubmit our request + */ + void get_version(string map, version_t *newest, version_t *oldest, Context *onfinish); + /** + * Run a callback within our lock, with a reference + * to the MonMap + */ + template<typename Callback, typename...Args> + auto with_monmap(Callback&& cb, Args&&...args) const -> + decltype(cb(monmap, std::forward<Args>(args)...)) { + std::lock_guard l(monc_lock); + return std::forward<Callback>(cb)(monmap, std::forward<Args>(args)...); + } + + void register_config_callback(md_config_t::config_callback fn); + void register_config_notify_callback(std::function<void(void)> f) { + config_notify_cb = f; + } + md_config_t::config_callback get_config_callback(); + +private: + struct version_req_d { + Context *context; + version_t *newest, *oldest; + version_req_d(Context *con, version_t *n, version_t *o) : context(con),newest(n), oldest(o) {} + }; + + map<ceph_tid_t, version_req_d*> version_requests; + ceph_tid_t version_req_id; + void handle_get_version_reply(MMonGetVersionReply* m); + + md_config_t::config_callback config_cb; + std::function<void(void)> config_notify_cb; +}; + +#endif |