+++ /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 <Python.h>
-
-#include "osdc/Objecter.h"
-#include "client/Client.h"
-#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 "mgr/mgr_commands.h"
-
-//#include "MgrPyModule.h"
-#include "DaemonServer.h"
-#include "messages/MMgrBeacon.h"
-#include "messages/MMgrDigest.h"
-#include "messages/MCommand.h"
-#include "messages/MCommandReply.h"
-#include "messages/MLog.h"
-#include "messages/MServiceMap.h"
-
-#include "Mgr.h"
-
-#define dout_context g_ceph_context
-#define dout_subsys ceph_subsys_mgr
-#undef dout_prefix
-#define dout_prefix *_dout << "mgr " << __func__ << " "
-
-
-Mgr::Mgr(MonClient *monc_, const MgrMap& mgrmap,
- PyModuleRegistry *py_module_registry_,
- Messenger *clientm_, Objecter *objecter_,
- Client* client_, LogChannelRef clog_, LogChannelRef audit_clog_) :
- monc(monc_),
- objecter(objecter_),
- client(client_),
- client_messenger(clientm_),
- lock("Mgr::lock"),
- timer(g_ceph_context, lock),
- finisher(g_ceph_context, "Mgr", "mgr-fin"),
- digest_received(false),
- py_module_registry(py_module_registry_),
- cluster_state(monc, nullptr, mgrmap),
- server(monc, finisher, daemon_state, cluster_state, *py_module_registry,
- clog_, audit_clog_),
- clog(clog_),
- audit_clog(audit_clog_),
- initialized(false),
- initializing(false)
-{
- cluster_state.set_objecter(objecter);
-}
-
-
-Mgr::~Mgr()
-{
-}
-
-
-/**
- * Context for completion of metadata mon commands: take
- * the result and stash it in DaemonStateIndex
- */
-class MetadataUpdate : public Context
-{
- DaemonStateIndex &daemon_state;
- DaemonKey key;
-
- std::map<std::string, std::string> defaults;
-
-public:
- bufferlist outbl;
- std::string outs;
-
- MetadataUpdate(DaemonStateIndex &daemon_state_, const DaemonKey &key_)
- : daemon_state(daemon_state_), key(key_) {}
-
- void set_default(const std::string &k, const std::string &v)
- {
- defaults[k] = v;
- }
-
- void finish(int r) override
- {
- daemon_state.clear_updating(key);
- if (r == 0) {
- if (key.first == "mds") {
- json_spirit::mValue json_result;
- bool read_ok = json_spirit::read(
- outbl.to_str(), json_result);
- if (!read_ok) {
- dout(1) << "mon returned invalid JSON for "
- << key.first << "." << key.second << dendl;
- return;
- }
-
- json_spirit::mObject daemon_meta = json_result.get_obj();
-
- // Apply any defaults
- for (const auto &i : defaults) {
- if (daemon_meta.find(i.first) == daemon_meta.end()) {
- daemon_meta[i.first] = i.second;
- }
- }
-
- DaemonStatePtr state;
- if (daemon_state.exists(key)) {
- state = daemon_state.get(key);
- Mutex::Locker l(state->lock);
- daemon_meta.erase("name");
- daemon_meta.erase("hostname");
- state->metadata.clear();
- for (const auto &i : daemon_meta) {
- state->metadata[i.first] = i.second.get_str();
- }
- } else {
- state = std::make_shared<DaemonState>(daemon_state.types);
- state->key = key;
- state->hostname = daemon_meta.at("hostname").get_str();
-
- for (const auto &i : daemon_meta) {
- state->metadata[i.first] = i.second.get_str();
- }
-
- daemon_state.insert(state);
- }
- } else if (key.first == "osd") {
- } else {
- ceph_abort();
- }
- } else {
- dout(1) << "mon failed to return metadata for "
- << key.first << "." << key.second << ": "
- << cpp_strerror(r) << dendl;
- }
- }
-};
-
-
-void Mgr::background_init(Context *completion)
-{
- Mutex::Locker l(lock);
- assert(!initializing);
- assert(!initialized);
- initializing = true;
-
- finisher.start();
-
- finisher.queue(new FunctionContext([this, completion](int r){
- init();
- completion->complete(0);
- }));
-}
-
-void Mgr::init()
-{
- Mutex::Locker l(lock);
- assert(initializing);
- assert(!initialized);
-
- // Start communicating with daemons to learn statistics etc
- int r = server.init(monc->get_global_id(), client_messenger->get_myaddr());
- if (r < 0) {
- derr << "Initialize server fail"<< dendl;
- return;
- }
- dout(4) << "Initialized server at " << server.get_myaddr() << dendl;
-
- // Preload all daemon metadata (will subsequently keep this
- // up to date by watching maps, so do the initial load before
- // we subscribe to any maps)
- dout(4) << "Loading daemon metadata..." << dendl;
- load_all_metadata();
-
- // subscribe to all the maps
- monc->sub_want("log-info", 0, 0);
- monc->sub_want("mgrdigest", 0, 0);
- monc->sub_want("fsmap", 0, 0);
- monc->sub_want("servicemap", 0, 0);
-
- dout(4) << "waiting for OSDMap..." << dendl;
- // Subscribe to OSDMap update to pass on to ClusterState
- objecter->maybe_request_map();
-
- // reset the mon session. we get these maps through subscriptions which
- // are stateful with the connection, so even if *we* don't have them a
- // previous incarnation sharing the same MonClient may have.
- monc->reopen_session();
-
- // Start Objecter and wait for OSD map
- lock.Unlock(); // Drop lock because OSDMap dispatch calls into my ms_dispatch
- objecter->wait_for_osd_map();
- lock.Lock();
-
- // Populate PGs in ClusterState
- objecter->with_osdmap([this](const OSDMap &osd_map) {
- cluster_state.notify_osdmap(osd_map);
- });
-
- // Wait for FSMap
- dout(4) << "waiting for FSMap..." << dendl;
- while (!cluster_state.have_fsmap()) {
- fs_map_cond.Wait(lock);
- }
-
- dout(4) << "waiting for config-keys..." << dendl;
-
- // Preload config keys (`get` for plugins is to be a fast local
- // operation, we we don't have to synchronize these later because
- // all sets will come via mgr)
- auto loaded_config = load_config();
-
- // Wait for MgrDigest...
- dout(4) << "waiting for MgrDigest..." << dendl;
- while (!digest_received) {
- digest_cond.Wait(lock);
- }
-
- // assume finisher already initialized in background_init
- dout(4) << "starting python modules..." << dendl;
- py_module_registry->active_start(loaded_config, daemon_state, cluster_state, *monc,
- clog, *objecter, *client, finisher);
-
- dout(4) << "Complete." << dendl;
- initializing = false;
- initialized = true;
-}
-
-void Mgr::load_all_metadata()
-{
- assert(lock.is_locked_by_me());
-
- JSONCommand mds_cmd;
- mds_cmd.run(monc, "{\"prefix\": \"mds metadata\"}");
- JSONCommand osd_cmd;
- osd_cmd.run(monc, "{\"prefix\": \"osd metadata\"}");
- JSONCommand mon_cmd;
- mon_cmd.run(monc, "{\"prefix\": \"mon metadata\"}");
-
- lock.Unlock();
- mds_cmd.wait();
- osd_cmd.wait();
- mon_cmd.wait();
- lock.Lock();
-
- assert(mds_cmd.r == 0);
- assert(mon_cmd.r == 0);
- assert(osd_cmd.r == 0);
-
- for (auto &metadata_val : mds_cmd.json_result.get_array()) {
- json_spirit::mObject daemon_meta = metadata_val.get_obj();
- if (daemon_meta.count("hostname") == 0) {
- dout(1) << "Skipping incomplete metadata entry" << dendl;
- continue;
- }
-
- DaemonStatePtr dm = std::make_shared<DaemonState>(daemon_state.types);
- dm->key = DaemonKey("mds",
- daemon_meta.at("name").get_str());
- dm->hostname = daemon_meta.at("hostname").get_str();
-
- daemon_meta.erase("name");
- daemon_meta.erase("hostname");
-
- for (const auto &i : daemon_meta) {
- dm->metadata[i.first] = i.second.get_str();
- }
-
- daemon_state.insert(dm);
- }
-
- for (auto &metadata_val : mon_cmd.json_result.get_array()) {
- json_spirit::mObject daemon_meta = metadata_val.get_obj();
- if (daemon_meta.count("hostname") == 0) {
- dout(1) << "Skipping incomplete metadata entry" << dendl;
- continue;
- }
-
- DaemonStatePtr dm = std::make_shared<DaemonState>(daemon_state.types);
- dm->key = DaemonKey("mon",
- daemon_meta.at("name").get_str());
- dm->hostname = daemon_meta.at("hostname").get_str();
-
- daemon_meta.erase("name");
- daemon_meta.erase("hostname");
-
- for (const auto &i : daemon_meta) {
- dm->metadata[i.first] = i.second.get_str();
- }
-
- daemon_state.insert(dm);
- }
-
- for (auto &osd_metadata_val : osd_cmd.json_result.get_array()) {
- json_spirit::mObject osd_metadata = osd_metadata_val.get_obj();
- if (osd_metadata.count("hostname") == 0) {
- dout(1) << "Skipping incomplete metadata entry" << dendl;
- continue;
- }
- dout(4) << osd_metadata.at("hostname").get_str() << dendl;
-
- DaemonStatePtr dm = std::make_shared<DaemonState>(daemon_state.types);
- dm->key = DaemonKey("osd",
- stringify(osd_metadata.at("id").get_int()));
- dm->hostname = osd_metadata.at("hostname").get_str();
-
- osd_metadata.erase("id");
- osd_metadata.erase("hostname");
-
- for (const auto &i : osd_metadata) {
- dm->metadata[i.first] = i.second.get_str();
- }
-
- daemon_state.insert(dm);
- }
-}
-
-std::map<std::string, std::string> Mgr::load_config()
-{
- assert(lock.is_locked_by_me());
-
- dout(10) << "listing keys" << dendl;
- JSONCommand cmd;
- cmd.run(monc, "{\"prefix\": \"config-key ls\"}");
- lock.Unlock();
- cmd.wait();
- lock.Lock();
- assert(cmd.r == 0);
-
- std::map<std::string, std::string> loaded;
-
- for (auto &key_str : cmd.json_result.get_array()) {
- std::string const key = key_str.get_str();
- dout(20) << "saw key '" << key << "'" << dendl;
-
- const std::string config_prefix = PyModuleRegistry::config_prefix;
-
- if (key.substr(0, config_prefix.size()) == config_prefix) {
- dout(20) << "fetching '" << key << "'" << dendl;
- Command get_cmd;
- std::ostringstream cmd_json;
- cmd_json << "{\"prefix\": \"config-key get\", \"key\": \"" << key << "\"}";
- get_cmd.run(monc, cmd_json.str());
- lock.Unlock();
- get_cmd.wait();
- lock.Lock();
- assert(get_cmd.r == 0);
- loaded[key] = get_cmd.outbl.to_str();
- }
- }
-
- return loaded;
-}
-
-void Mgr::shutdown()
-{
- finisher.queue(new FunctionContext([&](int) {
- {
- Mutex::Locker l(lock);
- monc->sub_unwant("log-info");
- monc->sub_unwant("mgrdigest");
- monc->sub_unwant("fsmap");
- // First stop the server so that we're not taking any more incoming
- // requests
- server.shutdown();
- }
- // after the messenger is stopped, signal modules to shutdown via finisher
- py_module_registry->active_shutdown();
- }));
-
- // Then stop the finisher to ensure its enqueued contexts aren't going
- // to touch references to the things we're about to tear down
- finisher.wait_for_empty();
- finisher.stop();
-}
-
-void Mgr::handle_osd_map()
-{
- assert(lock.is_locked_by_me());
-
- std::set<std::string> names_exist;
-
- /**
- * When we see a new OSD map, inspect the entity addrs to
- * see if they have changed (service restart), and if so
- * reload the metadata.
- */
- objecter->with_osdmap([this, &names_exist](const OSDMap &osd_map) {
- for (unsigned int osd_id = 0; osd_id < osd_map.get_num_osds(); ++osd_id) {
- if (!osd_map.exists(osd_id)) {
- continue;
- }
-
- // Remember which OSDs exist so that we can cull any that don't
- names_exist.insert(stringify(osd_id));
-
- // Consider whether to update the daemon metadata (new/restarted daemon)
- bool update_meta = false;
- const auto k = DaemonKey("osd", stringify(osd_id));
- if (daemon_state.is_updating(k)) {
- continue;
- }
-
- if (daemon_state.exists(k)) {
- auto metadata = daemon_state.get(k);
- Mutex::Locker l(metadata->lock);
- auto addr_iter = metadata->metadata.find("front_addr");
- if (addr_iter != metadata->metadata.end()) {
- const std::string &metadata_addr = addr_iter->second;
- const auto &map_addr = osd_map.get_addr(osd_id);
-
- if (metadata_addr != stringify(map_addr)) {
- dout(4) << "OSD[" << osd_id << "] addr change " << metadata_addr
- << " != " << stringify(map_addr) << dendl;
- update_meta = true;
- } else {
- dout(20) << "OSD[" << osd_id << "] addr unchanged: "
- << metadata_addr << dendl;
- }
- } else {
- // Awkward case where daemon went into DaemonState because it
- // sent us a report but its metadata didn't get loaded yet
- update_meta = true;
- }
- } else {
- update_meta = true;
- }
-
- if (update_meta) {
- daemon_state.notify_updating(k);
- auto c = new MetadataUpdate(daemon_state, k);
- std::ostringstream cmd;
- cmd << "{\"prefix\": \"osd metadata\", \"id\": "
- << osd_id << "}";
- monc->start_mon_command(
- {cmd.str()},
- {}, &c->outbl, &c->outs, c);
- }
- }
-
- cluster_state.notify_osdmap(osd_map);
- });
-
- // TODO: same culling for MonMap
- daemon_state.cull("osd", names_exist);
-}
-
-void Mgr::handle_log(MLog *m)
-{
- for (const auto &e : m->entries) {
- py_module_registry->notify_all(e);
- }
-
- m->put();
-}
-
-void Mgr::handle_service_map(MServiceMap *m)
-{
- dout(10) << "e" << m->service_map.epoch << dendl;
- cluster_state.set_service_map(m->service_map);
- server.got_service_map();
-}
-
-bool Mgr::ms_dispatch(Message *m)
-{
- dout(4) << *m << dendl;
- Mutex::Locker l(lock);
-
- switch (m->get_type()) {
- case MSG_MGR_DIGEST:
- handle_mgr_digest(static_cast<MMgrDigest*>(m));
- break;
- case CEPH_MSG_MON_MAP:
- py_module_registry->notify_all("mon_map", "");
- m->put();
- break;
- case CEPH_MSG_FS_MAP:
- py_module_registry->notify_all("fs_map", "");
- handle_fs_map((MFSMap*)m);
- return false; // I shall let this pass through for Client
- break;
- case CEPH_MSG_OSD_MAP:
- handle_osd_map();
-
- py_module_registry->notify_all("osd_map", "");
-
- // Continuous subscribe, so that we can generate notifications
- // for our MgrPyModules
- objecter->maybe_request_map();
- m->put();
- break;
- case MSG_SERVICE_MAP:
- handle_service_map((MServiceMap*)m);
- py_module_registry->notify_all("service_map", "");
- m->put();
- break;
- case MSG_LOG:
- handle_log(static_cast<MLog *>(m));
- break;
-
- default:
- return false;
- }
- return true;
-}
-
-
-void Mgr::handle_fs_map(MFSMap* m)
-{
- assert(lock.is_locked_by_me());
-
- std::set<std::string> names_exist;
-
- const FSMap &new_fsmap = m->get_fsmap();
-
- fs_map_cond.Signal();
-
- // TODO: callers (e.g. from python land) are potentially going to see
- // the new fsmap before we've bothered populating all the resulting
- // daemon_state. Maybe we should block python land while we're making
- // this kind of update?
-
- cluster_state.set_fsmap(new_fsmap);
-
- auto mds_info = new_fsmap.get_mds_info();
- for (const auto &i : mds_info) {
- const auto &info = i.second;
-
- if (!new_fsmap.gid_exists(i.first)){
- continue;
- }
-
- // Remember which MDS exists so that we can cull any that don't
- names_exist.insert(info.name);
-
- const auto k = DaemonKey("mds", info.name);
- if (daemon_state.is_updating(k)) {
- continue;
- }
-
- bool update = false;
- if (daemon_state.exists(k)) {
- auto metadata = daemon_state.get(k);
- Mutex::Locker l(metadata->lock);
- if (metadata->metadata.empty() ||
- metadata->metadata.count("addr") == 0) {
- update = true;
- } else {
- auto metadata_addr = metadata->metadata.at("addr");
- const auto map_addr = info.addr;
- update = metadata_addr != stringify(map_addr);
- if (update) {
- dout(4) << "MDS[" << info.name << "] addr change " << metadata_addr
- << " != " << stringify(map_addr) << dendl;
- }
- }
- } else {
- update = true;
- }
-
- if (update) {
- daemon_state.notify_updating(k);
- auto c = new MetadataUpdate(daemon_state, k);
-
- // Older MDS daemons don't have addr in the metadata, so
- // fake it if the returned metadata doesn't have the field.
- c->set_default("addr", stringify(info.addr));
-
- std::ostringstream cmd;
- cmd << "{\"prefix\": \"mds metadata\", \"who\": \""
- << info.name << "\"}";
- monc->start_mon_command(
- {cmd.str()},
- {}, &c->outbl, &c->outs, c);
- }
- }
- daemon_state.cull("mds", names_exist);
-}
-
-bool Mgr::got_mgr_map(const MgrMap& m)
-{
- Mutex::Locker l(lock);
- dout(10) << m << dendl;
-
- set<string> old_modules;
- cluster_state.with_mgrmap([&](const MgrMap& m) {
- old_modules = m.modules;
- });
- if (m.modules != old_modules) {
- derr << "mgrmap module list changed to (" << m.modules << "), respawn"
- << dendl;
- return true;
- }
-
- cluster_state.set_mgr_map(m);
-
- return false;
-}
-
-void Mgr::handle_mgr_digest(MMgrDigest* m)
-{
- dout(10) << m->mon_status_json.length() << dendl;
- dout(10) << m->health_json.length() << dendl;
- cluster_state.load_digest(m);
- py_module_registry->notify_all("mon_status", "");
- py_module_registry->notify_all("health", "");
-
- // Hack: use this as a tick/opportunity to prompt python-land that
- // the pgmap might have changed since last time we were here.
- py_module_registry->notify_all("pg_summary", "");
- dout(10) << "done." << dendl;
-
- m->put();
-
- if (!digest_received) {
- digest_received = true;
- digest_cond.Signal();
- }
-}
-
-void Mgr::tick()
-{
- dout(10) << dendl;
- server.send_report();
-}
-
-std::vector<MonCommand> Mgr::get_command_set() const
-{
- Mutex::Locker l(lock);
-
- std::vector<MonCommand> commands = mgr_commands;
- std::vector<MonCommand> py_commands = py_module_registry->get_commands();
- commands.insert(commands.end(), py_commands.begin(), py_commands.end());
- return commands;
-}
-
-std::map<std::string, std::string> Mgr::get_services() const
-{
- Mutex::Locker l(lock);
-
- return py_module_registry->get_services();
-}
-
Code Review / stor4nfv.git / blobdiff