diff options
Diffstat (limited to 'src/mon/MonClient.h')
| -rw-r--r-- | src/mon/MonClient.h | 774 |
1 files changed, 774 insertions, 0 deletions
diff --git a/src/mon/MonClient.h b/src/mon/MonClient.h new file mode 100644 index 000000000..6a7daa814 --- /dev/null +++ b/src/mon/MonClient.h @@ -0,0 +1,774 @@ +// -*- 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 <functional> +#include <list> +#include <map> +#include <memory> +#include <set> +#include <string> +#include <vector> + +#include "msg/Messenger.h" + +#include "MonMap.h" +#include "MonSub.h" + +#include "common/async/completion.h" +#include "common/Timer.h" +#include "common/config.h" +#include "messages/MMonGetVersion.h" + +#include "auth/AuthClient.h" +#include "auth/AuthServer.h" + +class MMonMap; +class MConfig; +class MMonGetVersionReply; +class MMonCommandAck; +class LogClient; +class AuthClientHandler; +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, + ceph::buffer::list *out, + const EntityName& entity_name, + uint32_t want_keys, + RotatingKeyRing* keyring); + int handle_auth_reply_more( + AuthConnectionMeta *auth_meta, + const ceph::buffer::list& bl, + ceph::buffer::list *reply); + int handle_auth_done( + AuthConnectionMeta *auth_meta, + uint64_t global_id, + const ceph::buffer::list& 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; + } + void queue_command(Message *m) { + pending_tell_command = m; + } + +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; + + MessageRef pending_tell_command; + + AuthRegistry *auth_registry; +}; + + +struct MonClientPinger : public Dispatcher, + public AuthClient { + ceph::mutex lock = ceph::make_mutex("MonClientPinger::lock"); + ceph::condition_variable ping_recvd_cond; + std::string *result; + bool done; + RotatingKeyRing *keyring; + std::unique_ptr<MonConnection> mc; + + MonClientPinger(CephContext *cct_, + RotatingKeyRing *keyring, + std::string *res_) : + Dispatcher(cct_), + result(res_), + done(false), + keyring(keyring) + { } + + int wait_for_reply(double timeout = 0.0) { + std::unique_lock locker{lock}; + if (timeout <= 0) { + timeout = std::chrono::duration<double>(cct->_conf.get_val<std::chrono::seconds>("client_mount_timeout")).count(); + } + done = false; + if (ping_recvd_cond.wait_for(locker, + ceph::make_timespan(timeout), + [this] { return done; })) { + return 0; + } else { + return ETIMEDOUT; + } + } + + bool ms_dispatch(Message *m) override { + using ceph::decode; + std::lock_guard l(lock); + if (m->get_type() != CEPH_MSG_PING) + return false; + + ceph::buffer::list &payload = m->get_payload(); + if (result && payload.length() > 0) { + auto p = std::cbegin(payload); + decode(*result, p); + } + done = true; + ping_recvd_cond.notify_all(); + m->put(); + return true; + } + bool ms_handle_reset(Connection *con) override { + std::lock_guard l(lock); + done = true; + ping_recvd_cond.notify_all(); + 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, + ceph::buffer::list *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 ceph::buffer::list& bl, + ceph::buffer::list *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 ceph::buffer::list& 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); + } +}; + +const boost::system::error_category& monc_category() noexcept; + +enum class monc_errc { + shutting_down = 1, // Command failed due to MonClient shutting down + session_reset, // Monitor session was reset + rank_dne, // Requested monitor rank does not exist + mon_dne, // Requested monitor does not exist + timed_out, // Monitor operation timed out + mon_unavailable // Monitor unavailable +}; + +namespace boost::system { +template<> +struct is_error_code_enum<::monc_errc> { + static const bool value = true; +}; +} + +// implicit conversion: +inline boost::system::error_code make_error_code(monc_errc e) noexcept { + return { static_cast<int>(e), monc_category() }; +} + +// explicit conversion: +inline boost::system::error_condition make_error_condition(monc_errc e) noexcept { + return { static_cast<int>(e), monc_category() }; +} + +const boost::system::error_category& monc_category() noexcept; + +class MonClient : public Dispatcher, + public AuthClient, + public AuthServer /* for mgr, osd, mds */ { + static constexpr auto dout_subsys = ceph_subsys_monc; +public: + // Error, Newest, Oldest + using VersionSig = void(boost::system::error_code, version_t, version_t); + using VersionCompletion = ceph::async::Completion<VersionSig>; + + using CommandSig = void(boost::system::error_code, std::string, + ceph::buffer::list); + using CommandCompletion = ceph::async::Completion<CommandSig>; + + MonMap monmap; + std::map<std::string,std::string> config_mgr; + +private: + Messenger *messenger; + + std::unique_ptr<MonConnection> active_con; + std::map<entity_addrvec_t, MonConnection> pending_cons; + std::set<unsigned> tried; + + EntityName entity_name; + + mutable ceph::mutex monc_lock = ceph::make_mutex("MonClient::monc_lock"); + SafeTimer timer; + boost::asio::io_context& service; + boost::asio::io_context::strand finish_strand{service}; + + 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 + utime_t last_keepalive; + utime_t last_send_log; + + void tick(); + void schedule_tick(); + + // monclient + bool want_monmap; + ceph::condition_variable map_cond; + bool passthrough_monmap = false; + + bool want_bootstrap_config = false; + ceph::ref_t<MConfig> bootstrap_config; + + // authenticate + std::unique_ptr<AuthClientHandler> auth; + uint32_t want_keys = 0; + uint64_t global_id = 0; + ceph::condition_variable auth_cond; + int authenticate_err = 0; + bool authenticated = false; + + std::list<MessageRef> waiting_for_session; + utime_t last_rotating_renew_sent; + 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); + void _add_conn(unsigned rank); + void _add_conns(); + void _un_backoff(); + void _send_mon_message(MessageRef 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, + ceph::buffer::list *bl) override; + int handle_auth_reply_more( + Connection *con, + AuthConnectionMeta *auth_meta, + const ceph::buffer::list& bl, + ceph::buffer::list *reply) override; + int handle_auth_done( + Connection *con, + AuthConnectionMeta *auth_meta, + uint64_t global_id, + uint32_t con_mode, + const ceph::buffer::list& 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 ceph::buffer::list& bl, + ceph::buffer::list *reply) override; + + 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(std::string what, version_t start, unsigned flags) { + std::lock_guard l(monc_lock); + return sub.want(what, start, flags); + } + void sub_got(std::string what, version_t have) { + std::lock_guard l(monc_lock); + sub.got(what, have); + } + void sub_unwant(std::string what) { + std::lock_guard l(monc_lock); + sub.unwant(what); + } + bool sub_want_increment(std::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: + MonClient(CephContext *cct_, boost::asio::io_context& service); + MonClient(const MonClient &) = delete; + MonClient& operator=(const MonClient &) = delete; + ~MonClient() override; + + int init(); + void shutdown(); + + void set_log_client(LogClient *clog) { + log_client = clog; + } + LogClient *get_log_client() { + return log_client; + } + + 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 std::string &mon_id, std::string *result_reply); + + void send_mon_message(Message *m) { + send_mon_message(MessageRef{m, false}); + } + void send_mon_message(MessageRef m); + + void reopen_session() { + std::lock_guard l(monc_lock); + _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 { + // for tell only + std::string target_name; + int target_rank = -1; + ConnectionRef target_con; + std::unique_ptr<MonConnection> target_session; + unsigned send_attempts = 0; ///< attempt count for legacy mons + utime_t last_send_attempt; + uint64_t tid; + std::vector<std::string> cmd; + ceph::buffer::list inbl; + std::unique_ptr<CommandCompletion> onfinish; + std::optional<boost::asio::steady_timer> cancel_timer; + + MonCommand(MonClient& monc, uint64_t t, std::unique_ptr<CommandCompletion> onfinish) + : tid(t), onfinish(std::move(onfinish)) { + auto timeout = + monc.cct->_conf.get_val<std::chrono::seconds>("rados_mon_op_timeout"); + if (timeout.count() > 0) { + cancel_timer.emplace(monc.service, timeout); + cancel_timer->async_wait( + [this, &monc](boost::system::error_code ec) { + if (ec) + return; + std::scoped_lock l(monc.monc_lock); + monc._cancel_mon_command(tid); + }); + } + } + + bool is_tell() const { + return target_name.size() || target_rank >= 0; + } + }; + friend MonCommand; + std::map<uint64_t,MonCommand*> mon_commands; + + void _send_command(MonCommand *r); + void _check_tell_commands(); + void _resend_mon_commands(); + int _cancel_mon_command(uint64_t tid); + void _finish_command(MonCommand *r, boost::system::error_code ret, std::string_view rs, + bufferlist&& bl); + void _finish_auth(); + void handle_mon_command_ack(MMonCommandAck *ack); + void handle_command_reply(MCommandReply *reply); + +public: + template<typename CompletionToken> + auto start_mon_command(const std::vector<std::string>& cmd, + const ceph::buffer::list& inbl, + CompletionToken&& token) { + ldout(cct,10) << __func__ << " cmd=" << cmd << dendl; + boost::asio::async_completion<CompletionToken, CommandSig> init(token); + { + std::scoped_lock l(monc_lock); + auto h = CommandCompletion::create(service.get_executor(), + std::move(init.completion_handler)); + if (!initialized || stopping) { + ceph::async::post(std::move(h), monc_errc::shutting_down, std::string{}, + bufferlist{}); + } else { + auto r = new MonCommand(*this, ++last_mon_command_tid, std::move(h)); + r->cmd = cmd; + r->inbl = inbl; + mon_commands.emplace(r->tid, r); + _send_command(r); + } + } + return init.result.get(); + } + + template<typename CompletionToken> + auto start_mon_command(int mon_rank, const std::vector<std::string>& cmd, + const ceph::buffer::list& inbl, CompletionToken&& token) { + ldout(cct,10) << __func__ << " cmd=" << cmd << dendl; + boost::asio::async_completion<CompletionToken, CommandSig> init(token); + { + std::scoped_lock l(monc_lock); + auto h = CommandCompletion::create(service.get_executor(), + std::move(init.completion_handler)); + if (!initialized || stopping) { + ceph::async::post(std::move(h), monc_errc::shutting_down, std::string{}, + bufferlist{}); + } else { + auto r = new MonCommand(*this, ++last_mon_command_tid, std::move(h)); + r->target_rank = mon_rank; + r->cmd = cmd; + r->inbl = inbl; + mon_commands.emplace(r->tid, r); + _send_command(r); + } + } + return init.result.get(); + } + + template<typename CompletionToken> + auto start_mon_command(const std::string& mon_name, + const std::vector<std::string>& cmd, + const ceph::buffer::list& inbl, + CompletionToken&& token) { + ldout(cct,10) << __func__ << " cmd=" << cmd << dendl; + boost::asio::async_completion<CompletionToken, CommandSig> init(token); + { + std::scoped_lock l(monc_lock); + auto h = CommandCompletion::create(service.get_executor(), + std::move(init.completion_handler)); + if (!initialized || stopping) { + ceph::async::post(std::move(h), monc_errc::shutting_down, std::string{}, + bufferlist{}); + } else { + auto r = new MonCommand(*this, ++last_mon_command_tid, std::move(h)); + // detect/tolerate mon *rank* passed as a string + std::string err; + int rank = strict_strtoll(mon_name.c_str(), 10, &err); + if (err.size() == 0 && rank >= 0) { + ldout(cct,10) << __func__ << " interpreting name '" << mon_name + << "' as rank " << rank << dendl; + r->target_rank = rank; + } else { + r->target_name = mon_name; + } + r->cmd = cmd; + r->inbl = inbl; + mon_commands.emplace(r->tid, r); + _send_command(r); + } + } + return init.result.get(); + } + + class ContextVerter { + std::string* outs; + ceph::bufferlist* outbl; + Context* onfinish; + + public: + ContextVerter(std::string* outs, ceph::bufferlist* outbl, Context* onfinish) + : outs(outs), outbl(outbl), onfinish(onfinish) {} + ~ContextVerter() = default; + ContextVerter(const ContextVerter&) = default; + ContextVerter& operator =(const ContextVerter&) = default; + ContextVerter(ContextVerter&&) = default; + ContextVerter& operator =(ContextVerter&&) = default; + + void operator()(boost::system::error_code e, + std::string s, + ceph::bufferlist bl) { + if (outs) + *outs = std::move(s); + if (outbl) + *outbl = std::move(bl); + if (onfinish) + onfinish->complete(ceph::from_error_code(e)); + } + }; + + void start_mon_command(const vector<string>& cmd, const bufferlist& inbl, + bufferlist *outbl, string *outs, + Context *onfinish) { + start_mon_command(cmd, inbl, ContextVerter(outs, outbl, onfinish)); + } + void start_mon_command(int mon_rank, + const vector<string>& cmd, const bufferlist& inbl, + bufferlist *outbl, string *outs, + Context *onfinish) { + start_mon_command(mon_rank, cmd, inbl, ContextVerter(outs, outbl, 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) { + start_mon_command(mon_name, cmd, inbl, ContextVerter(outs, outbl, onfinish)); + } + + + // version requests +public: + /** + * get latest known version(s) of cluster map + * + * @param map string name of map (e.g., 'osdmap') + * @param token context that will be triggered on completion + * @return (via Completion) {} on success, + * boost::system::errc::resource_unavailable_try_again if we need to + * resubmit our request + */ + template<typename CompletionToken> + auto get_version(std::string&& map, CompletionToken&& token) { + boost::asio::async_completion<CompletionToken, VersionSig> init(token); + { + std::scoped_lock l(monc_lock); + auto m = ceph::make_message<MMonGetVersion>(); + m->what = std::move(map); + m->handle = ++version_req_id; + version_requests.emplace(m->handle, + VersionCompletion::create( + service.get_executor(), + std::move(init.completion_handler))); + _send_mon_message(m); + } + return init.result.get(); + } + + /** + * 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: + + std::map<ceph_tid_t, std::unique_ptr<VersionCompletion>> 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 |