mgr/PyFormatter.cc
mgr/PyState.cc
mgr/MgrPyModule.cc
+ mgr/MgrStandby.cc
mgr/Mgr.cc)
add_executable(ceph-mgr ${mgr_srcs}
$<TARGET_OBJECTS:heap_profiler_objs>)
*
*/
-#include "mgr/Mgr.h"
#include "include/types.h"
#include "common/config.h"
#include "common/errno.h"
#include "global/global_init.h"
+#include "mgr/MgrStandby.h"
-
+/**
+ * A short main() which just instantiates a MgrStandby and
+ * hands over control to that.
+ */
int main(int argc, const char **argv)
{
vector<const char*> args;
global_init(NULL, args, CEPH_ENTITY_TYPE_MGR, CODE_ENVIRONMENT_DAEMON, 0,
"mgr_data");
- common_init_finish(g_ceph_context);
// For consumption by KeyRing::from_ceph_context in MonClient
g_conf->set_val("keyring", "$mgr_data/keyring", false);
- Mgr mgr;
+ MgrStandby mgr;
- // Handle --help before calling init() so we don't depend on network.
+ // Handle --help
if ((args.size() == 1 && (std::string(args[0]) == "--help" || std::string(args[0]) == "-h"))) {
mgr.usage();
return 0;
global_init_chdir(g_ceph_context);
common_init_finish(g_ceph_context);
- // Connect to mon cluster, download MDS map etc
int rc = mgr.init();
if (rc != 0) {
std::cerr << "Error in initialization: " << cpp_strerror(rc) << std::endl;
return rc;
}
- // Finally, execute the user's commands
return mgr.main(args);
}
OPTION(mgr_module_path, OPT_STR, CEPH_PKGLIBDIR "/mgr") // where to load python modules from
OPTION(mgr_modules, OPT_STR, "rest") // Which modules to load
OPTION(mgr_data, OPT_STR, "/var/lib/ceph/mgr/$cluster-$id") // where to find keyring etc
+OPTION(mgr_beacon_period, OPT_INT, 5) // How frequently to send beacon
+OPTION(mon_mgr_beacon_grace, OPT_INT, 30) // How long to wait to failover
OPTION(rgw_list_bucket_min_readahead, OPT_INT, 1000) // minimum number of entries to read from rados for bucket listing
protected:
uint64_t gid;
entity_addr_t server_addr;
+ bool available;
+ std::string name;
public:
MMgrBeacon()
- : PaxosServiceMessage(MSG_MGR_BEACON, 0, HEAD_VERSION, COMPAT_VERSION)
+ : PaxosServiceMessage(MSG_MGR_BEACON, 0, HEAD_VERSION, COMPAT_VERSION),
+ gid(0), available(false)
{
}
- MMgrBeacon(uint64_t gid_, entity_addr_t server_addr_)
+ MMgrBeacon(uint64_t gid_, const std::string &name_,
+ entity_addr_t server_addr_, bool available_)
: PaxosServiceMessage(MSG_MGR_BEACON, 0, HEAD_VERSION, COMPAT_VERSION),
- gid(gid_), server_addr(server_addr_)
+ gid(gid_), server_addr(server_addr_), available(available_), name(name_)
{
}
uint64_t get_gid() const { return gid; }
entity_addr_t get_server_addr() const { return server_addr; }
+ bool get_available() const { return available; }
+ const std::string& get_name() const { return name; }
private:
~MMgrBeacon() {}
const char *get_type_name() const { return "mgrbeacon"; }
void print(ostream& out) const {
- out << get_type_name() << "(" << gid << ", " << server_addr << ")";
+ out << get_type_name() << " mgr." << name << "(" << gid << ", "
+ << server_addr << ", " << available << ")";
}
void encode_payload(uint64_t features) {
paxos_encode();
::encode(server_addr, payload, features);
+ ::encode(gid, payload);
+ ::encode(available, payload);
+ ::encode(name, payload);
}
void decode_payload() {
bufferlist::iterator p = payload.begin();
paxos_decode(p);
::decode(server_addr, p);
+ ::decode(gid, p);
+ ::decode(available, p);
+ ::decode(name, p);
}
};
#define dout_prefix *_dout << "mgr " << __func__ << " "
-Mgr::Mgr() :
- Dispatcher(g_ceph_context),
+Mgr::Mgr(MonClient *monc_, Messenger *clientm_) :
+ monc(monc_),
objecter(NULL),
- monc(new MonClient(g_ceph_context)),
+ client_messenger(clientm_),
lock("Mgr::lock"),
timer(g_ceph_context, lock),
finisher(g_ceph_context, "Mgr", "mgr-fin"),
waiting_for_fs_map(NULL),
py_modules(daemon_state, cluster_state, *monc, finisher),
cluster_state(monc, nullptr),
- server(monc, daemon_state, py_modules)
+ server(monc, daemon_state, py_modules),
+ initialized(false),
+ initializing(false)
{
- client_messenger = Messenger::create_client_messenger(g_ceph_context, "mds");
-
- // FIXME: using objecter as convenience to handle incremental
- // OSD maps, but that's overkill. We don't really need an objecter.
- // Could we separate out the part of Objecter that we really need?
+ // Using Objecter to handle incremental decode of OSDMap
objecter = new Objecter(g_ceph_context, client_messenger, monc, NULL, 0, 0);
cluster_state.set_objecter(objecter);
Mgr::~Mgr()
{
delete objecter;
- delete monc;
- delete client_messenger;
- assert(waiting_for_fs_map == NULL);
+ assert(waiting_for_fs_map == nullptr);
}
};
-
-int Mgr::init()
+void Mgr::background_init()
{
Mutex::Locker l(lock);
+ assert(!initializing);
+ assert(!initialized);
+ initializing = true;
- // Initialize Messenger
- int r = client_messenger->bind(g_conf->public_addr);
- if (r < 0)
- return r;
+ finisher.start();
- client_messenger->start();
+ finisher.queue(new C_StdFunction([this](){
+ init();
+ }));
+}
+
+void Mgr::init()
+{
+ Mutex::Locker l(lock);
+ assert(initializing);
+ assert(!initialized);
objecter->set_client_incarnation(0);
objecter->init();
- // Connect dispatchers before starting objecter
- client_messenger->add_dispatcher_tail(objecter);
- client_messenger->add_dispatcher_tail(this);
-
- // Initialize MonClient
- if (monc->build_initial_monmap() < 0) {
- objecter->shutdown();
- client_messenger->shutdown();
- client_messenger->wait();
- return -1;
- }
-
- monc->set_want_keys(CEPH_ENTITY_TYPE_MON|CEPH_ENTITY_TYPE_OSD
- |CEPH_ENTITY_TYPE_MDS|CEPH_ENTITY_TYPE_MGR);
- monc->set_messenger(client_messenger);
- monc->init();
- r = monc->authenticate();
- if (r < 0) {
- derr << "Authentication failed, did you specify a mgr ID with a valid keyring?" << dendl;
- monc->shutdown();
- objecter->shutdown();
- client_messenger->shutdown();
- client_messenger->wait();
- return r;
- }
-
- client_t whoami = monc->get_global_id();
- client_messenger->set_myname(entity_name_t::CLIENT(whoami.v));
+ // Dispatcher before starting objecter
+ client_messenger->add_dispatcher_head(objecter);
// Start communicating with daemons to learn statistics etc
server.init(monc->get_global_id(), client_messenger->get_myaddr());
-
dout(4) << "Initialized server at " << server.get_myaddr() << dendl;
- // TODO: send the beacon periodically
- MMgrBeacon *m = new MMgrBeacon(monc->get_global_id(),
- server.get_myaddr());
- monc->send_mon_message(m);
// Preload all daemon metadata (will subsequently keep this
// up to date by watching maps, so do the initial load before
// Start Objecter and wait for OSD map
objecter->start();
+ lock.Unlock(); // Drop lock because OSDMap dispatch calls into my ms_dispatch
objecter->wait_for_osd_map();
- timer.init();
+ lock.Lock();
monc->sub_want("mgrdigest", 0, 0);
// Wait for MgrDigest...?
// TODO
- finisher.start();
+ // assume finisher already initialized in background_init
py_modules.init();
+ py_modules.start();
dout(4) << "Complete." << dendl;
- return 0;
+ initializing = false;
+ initialized = true;
}
void Mgr::load_all_metadata()
py_modules.insert_config(loaded);
}
-void Mgr::handle_signal(int signum)
+void Mgr::shutdown()
{
+ // FIXME: pre-empt init() if it is currently running, so that it will
+ // give up the lock for us.
Mutex::Locker l(lock);
- assert(signum == SIGINT || signum == SIGTERM);
- shutdown();
-}
-void Mgr::shutdown()
-{
// First stop the server so that we're not taking any more incoming requests
server.shutdown();
// to touch references to the things we're about to tear down
finisher.stop();
- //lock.Lock();
- timer.shutdown();
objecter->shutdown();
- //lock.Unlock();
-
- monc->shutdown();
- client_messenger->shutdown();
}
void Mgr::handle_osd_map()
}
-bool Mgr::ms_get_authorizer(int dest_type, AuthAuthorizer **authorizer,
- bool force_new)
-{
- if (dest_type == CEPH_ENTITY_TYPE_MON)
- return true;
-
- if (force_new) {
- if (monc->wait_auth_rotating(10) < 0)
- return false;
- }
-
- *authorizer = monc->auth->build_authorizer(dest_type);
- return *authorizer != NULL;
-}
-
-// A reference for use by the signal handler
-Mgr *signal_mgr = nullptr;
-
-static void handle_mgr_signal(int signum)
-{
- if (signal_mgr) {
- signal_mgr->handle_signal(signum);
- }
-}
-
-int Mgr::main(vector<const char *> args)
-{
- py_modules.start();
-
- // Enable signal handlers
- signal_mgr = this;
- init_async_signal_handler();
- register_async_signal_handler_oneshot(SIGINT, handle_mgr_signal);
- register_async_signal_handler_oneshot(SIGTERM, handle_mgr_signal);
-
- client_messenger->wait();
-
- // Disable signal handlers
- unregister_async_signal_handler(SIGINT, handle_mgr_signal);
- unregister_async_signal_handler(SIGTERM, handle_mgr_signal);
- shutdown_async_signal_handler();
- signal_mgr = nullptr;
-}
-
-
void Mgr::handle_mgr_digest(MMgrDigest* m)
{
dout(10) << m->mon_status_json.length() << dendl;
* Foundation. See file COPYING.
*/
-#ifndef CEPH_PYFOO_H_
-#define CEPH_PYFOO_H_
+#ifndef CEPH_MGR_H_
+#define CEPH_MGR_H_
// Python.h comes first because otherwise it clobbers ceph's assert
#include "Python.h"
#include "osdc/Objecter.h"
#include "mds/FSMap.h"
#include "messages/MFSMap.h"
-#include "msg/Dispatcher.h"
#include "msg/Messenger.h"
#include "auth/Auth.h"
#include "common/Finisher.h"
class MgrPyModule;
-class Mgr : public Dispatcher {
+class Mgr {
protected:
- Objecter *objecter;
MonClient *monc;
+ Objecter *objecter;
Messenger *client_messenger;
Mutex lock;
void load_config();
void load_all_metadata();
+ void init();
+
+ bool initialized;
+ bool initializing;
public:
- Mgr();
+ Mgr(MonClient *monc_, Messenger *clientm_);
~Mgr();
+ bool is_initialized() const {return initialized;}
+ entity_addr_t get_server_addr() const { return server.get_myaddr(); }
+
void handle_mgr_digest(MMgrDigest* m);
void handle_fs_map(MFSMap* m);
void handle_osd_map();
+
bool ms_dispatch(Message *m);
- bool ms_handle_reset(Connection *con) { return false; }
- void ms_handle_remote_reset(Connection *con) {}
- bool ms_get_authorizer(int dest_type, AuthAuthorizer **authorizer,
- bool force_new);
- int init();
+
+ void background_init();
void shutdown();
- void usage() {}
int main(vector<const char *> args);
- void handle_signal(int signum);
};
-#endif /* MDS_UTILITY_H_ */
+#endif
assert(msgr != nullptr);
timer.init();
-
-#if 0
- if (map.epoch == 0) {
- ldout(cct, 4) << "no map yet, waiting..." << dendl;
- wait_on_list(waiting_for_map);
- }
- ldout(cct, 4) << "proceeding with map " << map.epoch << dendl;
-#endif
}
void MgrClient::shutdown()
if (session == nullptr ||
session->con->get_peer_addr() != map.get_active_addr()) {
- entity_inst_t inst;
- inst.addr = map.get_active_addr();
- inst.name = entity_name_t::MGR(map.get_active_gid());
-
- delete session;
- session = new MgrSessionState();
- session->con = msgr->get_connection(inst);
-
- // Don't send an open if we're just a client (i.e. doing
- // command-sending, not stats etc)
- if (g_conf && !g_conf->name.is_client()) {
- auto open = new MMgrOpen();
- open->daemon_name = g_conf->name.get_id();
- session->con->send_message(open);
+ if (session) {
+ ldout(cct, 4) << "Terminating session with "
+ << session->con->get_peer_addr() << dendl;
+ delete session;
+ session = nullptr;
+
+ std::vector<ceph_tid_t> erase_cmds;
+ auto commands = command_table.get_commands();
+ for (const auto &i : commands) {
+ // FIXME be nicer, retarget command on new mgr?
+ if (i.second->on_finish != nullptr) {
+ i.second->on_finish->complete(-ETIMEDOUT);
+ }
+ erase_cmds.push_back(i.first);
+ }
+ for (const auto &tid : erase_cmds) {
+ command_table.erase(tid);
+ }
}
- std::vector<ceph_tid_t> erase_cmds;
- auto commands = command_table.get_commands();
- for (const auto &i : commands) {
- // FIXME be nicer, retarget command on new mgr?
- if (i.second->on_finish != nullptr) {
- i.second->on_finish->complete(-ETIMEDOUT);
+ if (map.get_available()) {
+ ldout(cct, 4) << "Starting new session with " << map.get_active_addr()
+ << dendl;
+ entity_inst_t inst;
+ inst.addr = map.get_active_addr();
+ inst.name = entity_name_t::MGR(map.get_active_gid());
+
+ session = new MgrSessionState();
+ session->con = msgr->get_connection(inst);
+
+ // Don't send an open if we're just a client (i.e. doing
+ // command-sending, not stats etc)
+ if (g_conf && !g_conf->name.is_client()) {
+ auto open = new MMgrOpen();
+ open->daemon_name = g_conf->name.get_id();
+ session->con->send_message(open);
}
- erase_cmds.push_back(i.first);
- }
- for (const auto &tid : erase_cmds) {
- command_table.erase(tid);
+
+ signal_cond_list(waiting_for_session);
+ } else {
+ ldout(cct, 4) << "No active mgr available yet" << dendl;
}
}
- signal_cond_list(waiting_for_map);
-
return true;
}
ldout(cct, 20) << "cmd: " << cmd << dendl;
- assert(map.epoch > 0);
-
-
-
if (session == nullptr) {
- derr << "no session" << dendl;
- // FIXME: be nicer: maybe block until a mgr is available?
- return -ENOENT;
+ derr << "no session, waiting" << dendl;
+ wait_on_list(waiting_for_session);
}
+ assert(map.epoch > 0);
+
MgrCommand *op = command_table.start_command();
op->cmd = cmd;
op->inbl = inbl;
void wait_on_list(list<Cond*>& ls);
void signal_cond_list(list<Cond*>& ls);
- list<Cond*> waiting_for_map;
+
+ list<Cond*> waiting_for_session;
public:
MgrClient(CephContext *cct_, Messenger *msgr_);
--- /dev/null
+// -*- 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) 2016 John Spray <john.spray@redhat.com>
+ *
+ * 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.
+ */
+
+#include "common/errno.h"
+#include "mon/MonClient.h"
+#include "include/stringify.h"
+#include "global/global_context.h"
+#include "global/signal_handler.h"
+
+#include "mgr/MgrContext.h"
+
+#include "messages/MMgrBeacon.h"
+#include "messages/MMgrMap.h"
+#include "Mgr.h"
+
+#include "MgrStandby.h"
+
+#define dout_subsys ceph_subsys_mgr
+#undef dout_prefix
+#define dout_prefix *_dout << "mgr " << __func__ << " "
+
+
+MgrStandby::MgrStandby() :
+ Dispatcher(g_ceph_context),
+ monc(new MonClient(g_ceph_context)),
+ lock("MgrStandby::lock"),
+ timer(g_ceph_context, lock),
+ active_mgr(nullptr)
+{
+ client_messenger = Messenger::create_client_messenger(g_ceph_context, "mgr");
+}
+
+
+MgrStandby::~MgrStandby()
+{
+ delete monc;
+ delete client_messenger;
+ delete active_mgr;
+}
+
+
+int MgrStandby::init()
+{
+ Mutex::Locker l(lock);
+
+ // Initialize Messenger
+ client_messenger->add_dispatcher_tail(this);
+ client_messenger->start();
+
+ // Initialize MonClient
+ if (monc->build_initial_monmap() < 0) {
+ client_messenger->shutdown();
+ client_messenger->wait();
+ return -1;
+ }
+
+ monc->sub_want("mgrmap", 0, 0);
+
+ monc->set_want_keys(CEPH_ENTITY_TYPE_MON|CEPH_ENTITY_TYPE_OSD
+ |CEPH_ENTITY_TYPE_MDS|CEPH_ENTITY_TYPE_MGR);
+ monc->set_messenger(client_messenger);
+ monc->init();
+ int r = monc->authenticate();
+ if (r < 0) {
+ derr << "Authentication failed, did you specify a mgr ID with a valid keyring?" << dendl;
+ monc->shutdown();
+ client_messenger->shutdown();
+ client_messenger->wait();
+ return r;
+ }
+
+ client_t whoami = monc->get_global_id();
+ client_messenger->set_myname(entity_name_t::CLIENT(whoami.v));
+
+ timer.init();
+ send_beacon();
+
+ dout(4) << "Complete." << dendl;
+ return 0;
+}
+
+void MgrStandby::send_beacon()
+{
+ assert(lock.is_locked_by_me());
+ dout(1) << state_str() << dendl;
+ dout(10) << "sending beacon as gid " << monc->get_global_id() << dendl;
+
+ bool available = active_mgr != nullptr && active_mgr->is_initialized();
+ auto addr = available ? active_mgr->get_server_addr() : entity_addr_t();
+ MMgrBeacon *m = new MMgrBeacon(monc->get_global_id(),
+ g_conf->name.get_id(),
+ addr,
+ available);
+
+ monc->send_mon_message(m);
+ // TODO configure period
+ timer.add_event_after(5, new C_StdFunction(
+ [this](){
+ send_beacon();
+ }
+ ));
+}
+
+void MgrStandby::handle_signal(int signum)
+{
+ Mutex::Locker l(lock);
+ assert(signum == SIGINT || signum == SIGTERM);
+ shutdown();
+}
+
+void MgrStandby::shutdown()
+{
+ // Expect already to be locked as we're called from signal handler
+ assert(lock.is_locked_by_me());
+
+ if (active_mgr) {
+ active_mgr->shutdown();
+ }
+
+ timer.shutdown();
+
+ monc->shutdown();
+ client_messenger->shutdown();
+}
+
+bool MgrStandby::ms_dispatch(Message *m)
+{
+ Mutex::Locker l(lock);
+ dout(4) << state_str() << " " << *m << dendl;
+
+ switch (m->get_type()) {
+ case MSG_MGR_MAP:
+ {
+ auto mmap = static_cast<MMgrMap*>(m);
+ auto map = mmap->get_map();
+ dout(4) << "received map epoch " << map.get_epoch() << dendl;
+ const bool active_in_map = map.active_gid == monc->get_global_id();
+ dout(4) << "active in map: " << active_in_map
+ << " active is " << map.active_gid << dendl;
+ if (active_in_map) {
+ if (active_mgr == nullptr) {
+ dout(1) << "Activating!" << dendl;
+ active_mgr = new Mgr(monc, client_messenger);
+ active_mgr->background_init();
+ dout(1) << "I am now active" << dendl;
+ } else {
+ dout(10) << "I was already active" << dendl;
+ }
+ } else {
+ if (active_mgr != nullptr) {
+ derr << "I was active but no longer am" << dendl;
+ shutdown();
+ // FIXME: should politely go back to standby (or respawn
+ // process) instead of stopping entirely.
+ }
+ }
+ }
+ break;
+
+ default:
+ if (active_mgr) {
+ return active_mgr->ms_dispatch(m);
+ } else {
+ return false;
+ }
+ }
+ return true;
+}
+
+
+bool MgrStandby::ms_get_authorizer(int dest_type, AuthAuthorizer **authorizer,
+ bool force_new)
+{
+ if (dest_type == CEPH_ENTITY_TYPE_MON)
+ return true;
+
+ if (force_new) {
+ if (monc->wait_auth_rotating(10) < 0)
+ return false;
+ }
+
+ *authorizer = monc->auth->build_authorizer(dest_type);
+ return *authorizer != NULL;
+}
+
+// A reference for use by the signal handler
+MgrStandby *signal_mgr = nullptr;
+
+static void handle_mgr_signal(int signum)
+{
+ if (signal_mgr) {
+ signal_mgr->handle_signal(signum);
+ }
+}
+
+int MgrStandby::main(vector<const char *> args)
+{
+ // Enable signal handlers
+ signal_mgr = this;
+ init_async_signal_handler();
+ register_async_signal_handler_oneshot(SIGINT, handle_mgr_signal);
+ register_async_signal_handler_oneshot(SIGTERM, handle_mgr_signal);
+
+ client_messenger->wait();
+
+ // Disable signal handlers
+ unregister_async_signal_handler(SIGINT, handle_mgr_signal);
+ unregister_async_signal_handler(SIGTERM, handle_mgr_signal);
+ shutdown_async_signal_handler();
+ signal_mgr = nullptr;
+
+ return 0;
+}
+
+
+std::string MgrStandby::state_str()
+{
+ return active_mgr == nullptr ? "standby" : "active";
+}
+
--- /dev/null
+// -*- 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) 2016 John Spray <john.spray@redhat.com>
+ *
+ * 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 MGR_STANDBY_H_
+#define MGR_STANDBY_H_
+
+#include "auth/Auth.h"
+#include "common/Finisher.h"
+#include "common/Timer.h"
+
+#include "DaemonServer.h"
+#include "PyModules.h"
+
+#include "DaemonState.h"
+#include "ClusterState.h"
+
+class Mgr;
+
+class MgrStandby : public Dispatcher {
+protected:
+ MonClient *monc;
+ Messenger *client_messenger;
+
+ Mutex lock;
+ SafeTimer timer;
+
+ Mgr *active_mgr;
+
+ std::string state_str();
+
+public:
+ MgrStandby();
+ ~MgrStandby();
+
+ bool ms_dispatch(Message *m);
+ bool ms_handle_reset(Connection *con) { return false; }
+ void ms_handle_remote_reset(Connection *con) {}
+ bool ms_get_authorizer(int dest_type, AuthAuthorizer **authorizer,
+ bool force_new);
+ int init();
+ void shutdown();
+ void usage() {}
+ int main(vector<const char *> args);
+ void handle_signal(int signum);
+ void send_beacon();
+};
+
+#endif
+
void PyModules::start()
{
- {
- Mutex::Locker l(lock);
- for (auto &i : modules) {
- auto thread = new ServeThread(i.second);
- serve_threads[i.first] = thread;
- }
+ Mutex::Locker l(lock);
+
+ dout(1) << "Creating threads for " << modules.size() << " modules" << dendl;
+ for (auto &i : modules) {
+ auto thread = new ServeThread(i.second);
+ serve_threads[i.first] = thread;
}
for (auto &i : serve_threads) {
#ifndef MGR_MAP_H_
#define MGR_MAP_H_
+#include <sstream>
+
#include "msg/msg_types.h"
+#include "common/Formatter.h"
#include "include/encoding.h"
+class StandbyInfo
+{
+public:
+ uint64_t gid;
+ std::string name;
+
+ StandbyInfo(uint64_t gid_, const std::string &name_)
+ : gid(gid_), name(name_)
+ {}
+
+ StandbyInfo()
+ : gid(0)
+ {}
+
+ void encode(bufferlist& bl) const
+ {
+ ENCODE_START(1, 1, bl);
+ ::encode(gid, bl);
+ ::encode(name, bl);
+ ENCODE_FINISH(bl);
+ }
+
+ void decode(bufferlist::iterator& p)
+ {
+ DECODE_START(1, p);
+ ::decode(gid, p);
+ ::decode(name, p);
+ DECODE_FINISH(p);
+ }
+};
+WRITE_CLASS_ENCODER(StandbyInfo)
+
class MgrMap
{
public:
+ epoch_t epoch;
+
+ // global_id of the ceph-mgr instance selected as a leader
uint64_t active_gid;
+ // server address reported by the leader once it is active
entity_addr_t active_addr;
- epoch_t epoch;
+ // whether the nominated leader is active (i.e. has initialized its server)
+ bool available;
+ // the name (foo in mgr.<foo>) of the active daemon
+ std::string active_name;
+
+ std::map<uint64_t, StandbyInfo> standbys;
epoch_t get_epoch() const { return epoch; }
entity_addr_t get_active_addr() const { return active_addr; }
uint64_t get_active_gid() const { return active_gid; }
+ bool get_available() const { return available; }
+ const std::string &get_active_name() const { return active_name; }
void encode(bufferlist& bl, uint64_t features) const
{
::encode(epoch, bl);
::encode(active_addr, bl, features);
::encode(active_gid, bl);
+ ::encode(available, bl);
+ ::encode(active_name, bl);
+ ::encode(standbys, bl);
ENCODE_FINISH(bl);
}
::decode(epoch, p);
::decode(active_addr, p);
::decode(active_gid, p);
+ ::decode(available, p);
+ ::decode(active_name, p);
+ ::decode(standbys, p);
DECODE_FINISH(p);
}
+ void print_summary(Formatter *f, std::ostream *ss) const
+ {
+ // One or the other, not both
+ assert((ss != nullptr) != (f != nullptr));
+
+ if (f) {
+ f->dump_int("active_gid", get_active_gid());
+ f->dump_string("active_name", get_active_name());
+ } else {
+ if (get_active_gid() != 0) {
+ *ss << "active: " << get_active_name() << " ";
+ } else {
+ *ss << "no daemons active ";
+ }
+ }
+
+
+ if (f) {
+ f->open_array_section("standbys");
+ for (const auto &i : standbys) {
+ f->open_object_section("standby");
+ f->dump_int("gid", i.second.gid);
+ f->dump_string("name", i.second.name);
+ f->close_section();
+ }
+ f->close_section();
+ } else {
+ if (standbys.size()) {
+ *ss << "standbys: ";
+ bool first = true;
+ for (const auto &i : standbys) {
+ if (!first) {
+ *ss << ", ";
+ }
+ *ss << i.second.name;
+ first = false;
+ }
+ }
+ }
+ }
+
MgrMap()
- : epoch(0)
+ : epoch(0), available(false)
{}
};
#include "PGMap.h"
#include "PGMonitor.h"
#include "include/stringify.h"
+#include "mgr/MgrContext.h"
#include "MgrMonitor.h"
switch (m->get_type()) {
case MSG_MGR_BEACON:
return preprocess_beacon(op);
+ case MSG_MON_COMMAND:
+ return preprocess_command(op);
default:
mon->no_reply(op);
derr << "Unhandled message type " << m->get_type() << dendl;
switch (m->get_type()) {
case MSG_MGR_BEACON:
return prepare_beacon(op);
+
+ case MSG_MON_COMMAND:
+ return prepare_command(op);
+
default:
mon->no_reply(op);
derr << "Unhandled message type " << m->get_type() << dendl;
bool MgrMonitor::prepare_beacon(MonOpRequestRef op)
{
MMgrBeacon *m = static_cast<MMgrBeacon*>(op->get_req());
+ dout(4) << "beacon from " << m->get_gid() << dendl;
+
+ // See if we are seeing same name, new GID for the active daemon
+ if (m->get_name() == pending_map.active_name
+ && m->get_gid() != pending_map.active_gid)
+ {
+ dout(4) << "Active daemon restart (mgr." << m->get_name() << ")" << dendl;
+ drop_active();
+ }
+
+ // See if we are seeing same name, new GID for any standbys
+ for (const auto &i : pending_map.standbys) {
+ const StandbyInfo &s = i.second;
+ if (s.name == m->get_name() && s.gid != m->get_gid()) {
+ dout(4) << "Standby daemon restart (mgr." << m->get_name() << ")" << dendl;
+ drop_standby(i.first);
+ break;
+ }
+ }
+
+ last_beacon[m->get_gid()] = ceph_clock_now(g_ceph_context);
+
+ // Track whether we modified pending_map
+ bool updated = false;
+
+ if (pending_map.active_gid == m->get_gid()) {
+ // A beacon from the currently active daemon
+ if (pending_map.active_addr != m->get_server_addr()) {
+ dout(4) << "learned address " << m->get_server_addr() << dendl;
+ pending_map.active_addr = m->get_server_addr();
+ updated = true;
+ }
+
+ if (pending_map.get_available() != m->get_available()) {
+ dout(4) << "available " << m->get_gid() << dendl;
+ pending_map.available = m->get_available();
+ updated = true;
+ }
+ } else if (pending_map.active_gid == 0) {
+ // There is no currently active daemon, select this one.
+ if (pending_map.standbys.count(m->get_gid())) {
+ drop_standby(m->get_gid());
+ }
+ pending_map.active_gid = m->get_gid();
+ pending_map.active_name = m->get_name();
- pending_map.active_gid = m->get_gid();
- pending_map.active_addr = m->get_server_addr();
+ dout(4) << "selecting new active in epoch " << pending_map.epoch << dendl;
+ wait_for_finished_proposal(op, new C_Updated(this, op));
+ } else {
+ if (pending_map.standbys.count(m->get_gid()) > 0) {
+ dout(10) << "from existing standby " << m->get_gid() << dendl;
+ } else {
+ dout(10) << "new standby " << m->get_gid() << dendl;
+ pending_map.standbys[m->get_gid()] = {m->get_gid(), m->get_name()};
+ updated = true;
+ }
+ }
- dout(4) << "proposing epoch " << pending_map.epoch << dendl;
- wait_for_finished_proposal(op, new C_Updated(this, op));
+ if (updated) {
+ dout(4) << "updating map" << dendl;
+ wait_for_finished_proposal(op, new C_Updated(this, op));
+ } else {
+ dout(10) << "no change" << dendl;
+ }
return true;
}
*/
void MgrMonitor::send_digests()
{
+ digest_callback = nullptr;
+
const std::string type = "mgrdigest";
if (mon->session_map.subs.count(type) == 0)
return;
sub->session->con->send_message(mdigest);
}
+
+ digest_callback = new C_StdFunction([this](){
+ send_digests();
+ });
+ mon->timer.add_event_after(5, digest_callback);
}
void MgrMonitor::tick()
{
- // TODO control frequency independently of the global tick frequency
- send_digests();
+ const utime_t now = ceph_clock_now(g_ceph_context);
+ utime_t cutoff = now;
+ cutoff -= g_conf->mon_mgr_beacon_grace;
+
+ // Populate any missing beacons (i.e. no beacon since MgrMonitor
+ // instantiation) with the current time, so that they will
+ // eventually look laggy if they fail to give us a beacon.
+ if (pending_map.active_gid != 0
+ && last_beacon.count(pending_map.active_gid) == 0) {
+ last_beacon[pending_map.active_gid] = now;
+ }
+ for (auto s : pending_map.standbys) {
+ if (last_beacon.count(s.first) == 0) {
+ last_beacon[s.first] = now;
+ }
+ }
+
+ // Cull standbys first so that any remaining standbys
+ // will be eligible to take over from the active if we cull him.
+ std::list<uint64_t> dead_standbys;
+ for (const auto &i : pending_map.standbys) {
+ auto last_beacon_time = last_beacon.at(i.first);
+ if (last_beacon_time < cutoff) {
+ dead_standbys.push_back(i.first);
+ }
+ }
+
+ bool propose = false;
+
+ for (auto i : dead_standbys) {
+ dout(4) << "Dropping laggy standby " << i << dendl;
+ drop_standby(i);
+ propose = true;
+ }
+
+ if (pending_map.active_gid != 0
+ && last_beacon.at(pending_map.active_gid) < cutoff) {
+
+ drop_active();
+ dout(4) << "Dropping active" << pending_map.active_gid << dendl;
+ if (promote_standby()) {
+ dout(4) << "Promoted standby " << pending_map.active_gid << dendl;
+ propose = true;
+ } else {
+ dout(4) << "Active is laggy but have no standbys to replace it" << dendl;
+ }
+ } else if (pending_map.active_gid == 0) {
+ if (promote_standby()) {
+ dout(4) << "Promoted standby " << pending_map.active_gid << dendl;
+ propose = true;
+ }
+ }
+
+ if (propose) {
+ propose_pending();
+ }
+}
+
+bool MgrMonitor::promote_standby()
+{
+ assert(pending_map.active_gid == 0);
+ if (pending_map.standbys.size()) {
+ // Promote a replacement (arbitrary choice of standby)
+ auto replacement_gid = pending_map.standbys.begin()->first;
+ pending_map.active_gid = replacement_gid;
+ pending_map.active_name = pending_map.standbys.at(replacement_gid).name;
+ pending_map.available = false;
+ pending_map.active_addr = entity_addr_t();
+
+ drop_standby(replacement_gid);
+ return true;
+ } else {
+ return false;
+ }
+}
+
+void MgrMonitor::drop_active()
+{
+ if (last_beacon.count(pending_map.active_gid) > 0) {
+ last_beacon.erase(pending_map.active_gid);
+ }
+
+ pending_map.active_name = "";
+ pending_map.active_gid = 0;
+ pending_map.available = false;
+ pending_map.active_addr = entity_addr_t();
+}
+
+void MgrMonitor::drop_standby(uint64_t gid)
+{
+ pending_map.standbys.erase(gid);
+ if (last_beacon.count(gid) > 0) {
+ last_beacon.erase(gid);
+ }
+
+}
+
+bool MgrMonitor::preprocess_command(MonOpRequestRef op)
+{
+ return false;
+
+}
+
+bool MgrMonitor::prepare_command(MonOpRequestRef op)
+{
+ MMonCommand *m = static_cast<MMonCommand*>(op->get_req());
+
+ std::stringstream ss;
+ bufferlist rdata;
+
+ std::map<std::string, cmd_vartype> cmdmap;
+ if (!cmdmap_from_json(m->cmd, &cmdmap, ss)) {
+ string rs = ss.str();
+ mon->reply_command(op, -EINVAL, rs, rdata, get_last_committed());
+ return true;
+ }
+
+ MonSession *session = m->get_session();
+ if (!session) {
+ mon->reply_command(op, -EACCES, "access denied", rdata, get_last_committed());
+ return true;
+ }
+
+ string prefix;
+ cmd_getval(g_ceph_context, cmdmap, "prefix", prefix);
+
+ int r = 0;
+
+ if (prefix == "mgr fail") {
+ string who;
+ cmd_getval(g_ceph_context, cmdmap, "who", who);
+
+ std::string err;
+ uint64_t gid = strict_strtol(who.c_str(), 10, &err);
+ bool changed = false;
+ if (!err.empty()) {
+ // Does not parse as a gid, treat it as a name
+ if (pending_map.active_name == who) {
+ drop_active();
+ changed = true;
+ } else {
+ gid = 0;
+ for (const auto &i : pending_map.standbys) {
+ if (i.second.name == who) {
+ gid = i.first;
+ }
+ }
+ if (gid != 0) {
+ drop_standby(gid);
+ changed = true;
+ } else {
+ ss << "Daemon not found '" << who << "', already failed?";
+ }
+ }
+ } else {
+ if (pending_map.active_gid == gid) {
+ drop_active();
+ changed = true;
+ } else if (pending_map.standbys.count(gid) > 0) {
+ drop_standby(gid);
+ changed = true;
+ } else {
+ ss << "Daemon not found '" << gid << "', already failed?";
+ }
+ }
+
+ if (changed) {
+ tick();
+ }
+ } else {
+ r = -ENOSYS;
+ }
+
+ dout(4) << __func__ << " done, r=" << r << dendl;
+ /* Compose response */
+ string rs;
+ getline(ss, rs);
+
+ if (r >= 0) {
+ // success.. delay reply
+ wait_for_finished_proposal(op, new Monitor::C_Command(mon, op, r, rs,
+ get_last_committed() + 1));
+ return true;
+ } else {
+ // reply immediately
+ mon->reply_command(op, r, rs, rdata, get_last_committed());
+ return false;
+ }
+}
+
+void MgrMonitor::init()
+{
+ send_digests(); // To get it to schedule its own event
+}
+
+void MgrMonitor::on_shutdown()
+{
+ if (digest_callback) {
+ mon->timer.cancel_event(digest_callback);
+ }
}
*/
+#include "include/Context.h"
#include "MgrMap.h"
#include "PaxosService.h"
+
class MgrMonitor : public PaxosService
{
MgrMap map;
MgrMap pending_map;
+ std::map<uint64_t, utime_t> last_beacon;
+
+ /**
+ * If a standby is available, make it active, given that
+ * there is currently no active daemon.
+ *
+ * @return true if a standby was promoted
+ */
+ bool promote_standby();
+ void drop_active();
+ void drop_standby(uint64_t gid);
+
+ Context *digest_callback;
+
public:
MgrMonitor(Monitor *mn, Paxos *p, const string& service_name)
- : PaxosService(mn, p, service_name)
+ : PaxosService(mn, p, service_name), digest_callback(nullptr)
{}
+ void init();
+ void on_shutdown();
+
+ const MgrMap &get_map() const { return map; }
+
+ bool in_use() const { return map.epoch > 0; }
+
void create_initial();
void update_from_paxos(bool *need_bootstrap);
void create_pending();
void encode_pending(MonitorDBStore::TransactionRef t);
+
bool preprocess_query(MonOpRequestRef op);
bool prepare_update(MonOpRequestRef op);
+
+ bool preprocess_command(MonOpRequestRef op);
+ bool prepare_command(MonOpRequestRef op);
+
void encode_full(MonitorDBStore::TransactionRef t) { }
bool preprocess_beacon(MonOpRequestRef op);
void tick();
+ void print_summary(Formatter *f, std::ostream *ss) const;
+
friend class C_Updated;
};
"name=key,type=CephString", \
"check for <key>'s existence", "config-key", "r", "cli,rest")
COMMAND("config-key list ", "list keys", "config-key", "r", "cli,rest")
+
+
+/*
+ * mon/MgrMonitor.cc
+ */
+COMMAND("mgr fail name=who,type=CephString", \
+ "treat the named manager daemon as failed", "mgr", "rw", "cli,rest")
f->open_object_section("fsmap");
mdsmon()->fsmap.print_summary(f, NULL);
f->close_section();
+
+ f->open_object_section("mgrmap");
+ mgrmon()->get_map().print_summary(f, nullptr);
+ f->close_section();
f->close_section();
} else {
ss << " cluster " << monmap->get_fsid() << "\n";
if (mdsmon()->fsmap.any_filesystems()) {
ss << " fsmap " << mdsmon()->fsmap << "\n";
}
+ if (mgrmon()->in_use()) {
+ ss << " mgr ";
+ mgrmon()->get_map().print_summary(nullptr, &ss);
+ ss << "\n";
+ }
osdmon()->osdmap.print_summary(NULL, ss);
pgmon()->pg_map.print_summary(NULL, &ss);
return;
}
+ if (module == "mgr") {
+ mgrmon()->dispatch(op);
+ return;
+ }
+
if (prefix == "fsid") {
if (f) {
f->open_object_section("fsid");
projects / ceph.git / commitdiff