diff options
Diffstat (limited to '')
| -rw-r--r-- | src/mon/MgrStatMonitor.cc | 367 |
1 files changed, 367 insertions, 0 deletions
diff --git a/src/mon/MgrStatMonitor.cc b/src/mon/MgrStatMonitor.cc new file mode 100644 index 000000000..9da4c50da --- /dev/null +++ b/src/mon/MgrStatMonitor.cc @@ -0,0 +1,367 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "MgrStatMonitor.h" +#include "mon/OSDMonitor.h" +#include "mon/MgrMonitor.h" +#include "mon/PGMap.h" +#include "messages/MGetPoolStats.h" +#include "messages/MGetPoolStatsReply.h" +#include "messages/MMonMgrReport.h" +#include "messages/MStatfs.h" +#include "messages/MStatfsReply.h" +#include "messages/MServiceMap.h" + +#include "include/ceph_assert.h" // re-clobber assert + +#define dout_subsys ceph_subsys_mon +#undef dout_prefix +#define dout_prefix _prefix(_dout, mon) + +using std::dec; +using std::hex; +using std::list; +using std::map; +using std::make_pair; +using std::ostream; +using std::ostringstream; +using std::pair; +using std::set; +using std::string; +using std::stringstream; +using std::to_string; +using std::vector; + +using ceph::bufferlist; +using ceph::decode; +using ceph::encode; +using ceph::ErasureCodeInterfaceRef; +using ceph::ErasureCodeProfile; +using ceph::Formatter; +using ceph::JSONFormatter; +using ceph::make_message; +using ceph::mono_clock; +using ceph::mono_time; + +static ostream& _prefix(std::ostream *_dout, Monitor &mon) { + return *_dout << "mon." << mon.name << "@" << mon.rank + << "(" << mon.get_state_name() + << ").mgrstat "; +} + +MgrStatMonitor::MgrStatMonitor(Monitor &mn, Paxos &p, const string& service_name) + : PaxosService(mn, p, service_name) +{ +} + +MgrStatMonitor::~MgrStatMonitor() = default; + +void MgrStatMonitor::create_initial() +{ + dout(10) << __func__ << dendl; + version = 0; + service_map.epoch = 1; + service_map.modified = ceph_clock_now(); + pending_service_map_bl.clear(); + encode(service_map, pending_service_map_bl, CEPH_FEATURES_ALL); +} + +void MgrStatMonitor::update_from_paxos(bool *need_bootstrap) +{ + version = get_last_committed(); + dout(10) << " " << version << dendl; + load_health(); + bufferlist bl; + get_version(version, bl); + if (version) { + ceph_assert(bl.length()); + try { + auto p = bl.cbegin(); + decode(digest, p); + decode(service_map, p); + if (!p.end()) { + decode(progress_events, p); + } + dout(10) << __func__ << " v" << version + << " service_map e" << service_map.epoch + << " " << progress_events.size() << " progress events" + << dendl; + } + catch (ceph::buffer::error& e) { + derr << "failed to decode mgrstat state; luminous dev version? " + << e.what() << dendl; + } + } + check_subs(); + update_logger(); + mon.osdmon()->notify_new_pg_digest(); +} + +void MgrStatMonitor::update_logger() +{ + dout(20) << __func__ << dendl; + + mon.cluster_logger->set(l_cluster_osd_bytes, digest.osd_sum.statfs.total); + mon.cluster_logger->set(l_cluster_osd_bytes_used, + digest.osd_sum.statfs.get_used_raw()); + mon.cluster_logger->set(l_cluster_osd_bytes_avail, + digest.osd_sum.statfs.available); + + mon.cluster_logger->set(l_cluster_num_pool, digest.pg_pool_sum.size()); + uint64_t num_pg = 0; + for (auto i : digest.num_pg_by_pool) { + num_pg += i.second; + } + mon.cluster_logger->set(l_cluster_num_pg, num_pg); + + unsigned active = 0, active_clean = 0, peering = 0; + for (auto p = digest.num_pg_by_state.begin(); + p != digest.num_pg_by_state.end(); + ++p) { + if (p->first & PG_STATE_ACTIVE) { + active += p->second; + if (p->first & PG_STATE_CLEAN) + active_clean += p->second; + } + if (p->first & PG_STATE_PEERING) + peering += p->second; + } + mon.cluster_logger->set(l_cluster_num_pg_active_clean, active_clean); + mon.cluster_logger->set(l_cluster_num_pg_active, active); + mon.cluster_logger->set(l_cluster_num_pg_peering, peering); + + mon.cluster_logger->set(l_cluster_num_object, digest.pg_sum.stats.sum.num_objects); + mon.cluster_logger->set(l_cluster_num_object_degraded, digest.pg_sum.stats.sum.num_objects_degraded); + mon.cluster_logger->set(l_cluster_num_object_misplaced, digest.pg_sum.stats.sum.num_objects_misplaced); + mon.cluster_logger->set(l_cluster_num_object_unfound, digest.pg_sum.stats.sum.num_objects_unfound); + mon.cluster_logger->set(l_cluster_num_bytes, digest.pg_sum.stats.sum.num_bytes); + +} + +void MgrStatMonitor::create_pending() +{ + dout(10) << " " << version << dendl; + pending_digest = digest; + pending_health_checks = get_health_checks(); + pending_service_map_bl.clear(); + encode(service_map, pending_service_map_bl, mon.get_quorum_con_features()); +} + +void MgrStatMonitor::encode_pending(MonitorDBStore::TransactionRef t) +{ + ++version; + dout(10) << " " << version << dendl; + bufferlist bl; + encode(pending_digest, bl, mon.get_quorum_con_features()); + ceph_assert(pending_service_map_bl.length()); + bl.append(pending_service_map_bl); + encode(pending_progress_events, bl); + put_version(t, version, bl); + put_last_committed(t, version); + + encode_health(pending_health_checks, t); +} + +version_t MgrStatMonitor::get_trim_to() const +{ + // we don't actually need *any* old states, but keep a few. + if (version > 5) { + return version - 5; + } + return 0; +} + +void MgrStatMonitor::on_active() +{ + update_logger(); +} + +void MgrStatMonitor::tick() +{ +} + +bool MgrStatMonitor::preprocess_query(MonOpRequestRef op) +{ + auto m = op->get_req<PaxosServiceMessage>(); + switch (m->get_type()) { + case CEPH_MSG_STATFS: + return preprocess_statfs(op); + case MSG_MON_MGR_REPORT: + return preprocess_report(op); + case MSG_GETPOOLSTATS: + return preprocess_getpoolstats(op); + default: + mon.no_reply(op); + derr << "Unhandled message type " << m->get_type() << dendl; + return true; + } +} + +bool MgrStatMonitor::prepare_update(MonOpRequestRef op) +{ + auto m = op->get_req<PaxosServiceMessage>(); + switch (m->get_type()) { + case MSG_MON_MGR_REPORT: + return prepare_report(op); + default: + mon.no_reply(op); + derr << "Unhandled message type " << m->get_type() << dendl; + return true; + } +} + +bool MgrStatMonitor::preprocess_report(MonOpRequestRef op) +{ + auto m = op->get_req<MMonMgrReport>(); + mon.no_reply(op); + if (m->gid && + m->gid != mon.mgrmon()->get_map().get_active_gid()) { + dout(10) << "ignoring report from non-active mgr " << m->gid + << dendl; + return true; + } + return false; +} + +bool MgrStatMonitor::prepare_report(MonOpRequestRef op) +{ + auto m = op->get_req<MMonMgrReport>(); + bufferlist bl = m->get_data(); + auto p = bl.cbegin(); + decode(pending_digest, p); + pending_health_checks.swap(m->health_checks); + if (m->service_map_bl.length()) { + pending_service_map_bl.swap(m->service_map_bl); + } + pending_progress_events.swap(m->progress_events); + dout(10) << __func__ << " " << pending_digest << ", " + << pending_health_checks.checks.size() << " health checks, " + << progress_events.size() << " progress events" << dendl; + dout(20) << "pending_digest:\n"; + JSONFormatter jf(true); + jf.open_object_section("pending_digest"); + pending_digest.dump(&jf); + jf.close_section(); + jf.flush(*_dout); + *_dout << dendl; + dout(20) << "health checks:\n"; + JSONFormatter jf(true); + jf.open_object_section("health_checks"); + pending_health_checks.dump(&jf); + jf.close_section(); + jf.flush(*_dout); + *_dout << dendl; + dout(20) << "progress events:\n"; + JSONFormatter jf(true); + jf.open_object_section("progress_events"); + for (auto& i : pending_progress_events) { + jf.dump_object(i.first.c_str(), i.second); + } + jf.close_section(); + jf.flush(*_dout); + *_dout << dendl; + return true; +} + +bool MgrStatMonitor::preprocess_getpoolstats(MonOpRequestRef op) +{ + op->mark_pgmon_event(__func__); + auto m = op->get_req<MGetPoolStats>(); + auto session = op->get_session(); + if (!session) + return true; + if (!session->is_capable("pg", MON_CAP_R)) { + dout(0) << "MGetPoolStats received from entity with insufficient caps " + << session->caps << dendl; + return true; + } + if (m->fsid != mon.monmap->fsid) { + dout(0) << __func__ << " on fsid " + << m->fsid << " != " << mon.monmap->fsid << dendl; + return true; + } + epoch_t ver = get_last_committed(); + auto reply = new MGetPoolStatsReply(m->fsid, m->get_tid(), ver); + reply->per_pool = digest.use_per_pool_stats(); + for (const auto& pool_name : m->pools) { + const auto pool_id = mon.osdmon()->osdmap.lookup_pg_pool_name(pool_name); + if (pool_id == -ENOENT) + continue; + auto pool_stat = get_pool_stat(pool_id); + if (!pool_stat) + continue; + reply->pool_stats[pool_name] = *pool_stat; + } + mon.send_reply(op, reply); + return true; +} + +bool MgrStatMonitor::preprocess_statfs(MonOpRequestRef op) +{ + op->mark_pgmon_event(__func__); + auto statfs = op->get_req<MStatfs>(); + auto session = op->get_session(); + + if (!session) + return true; + if (!session->is_capable("pg", MON_CAP_R)) { + dout(0) << "MStatfs received from entity with insufficient privileges " + << session->caps << dendl; + return true; + } + if (statfs->fsid != mon.monmap->fsid) { + dout(0) << __func__ << " on fsid " << statfs->fsid + << " != " << mon.monmap->fsid << dendl; + return true; + } + const auto& pool = statfs->data_pool; + if (pool && !mon.osdmon()->osdmap.have_pg_pool(*pool)) { + // There's no error field for MStatfsReply so just ignore the request. + // This is known to happen when a client is still accessing a removed fs. + dout(1) << __func__ << " on removed pool " << *pool << dendl; + return true; + } + dout(10) << __func__ << " " << *statfs + << " from " << statfs->get_orig_source() << dendl; + epoch_t ver = get_last_committed(); + auto reply = new MStatfsReply(statfs->fsid, statfs->get_tid(), ver); + reply->h.st = get_statfs(mon.osdmon()->osdmap, pool); + mon.send_reply(op, reply); + return true; +} + +void MgrStatMonitor::check_sub(Subscription *sub) +{ + dout(10) << __func__ + << " next " << sub->next + << " vs service_map.epoch " << service_map.epoch << dendl; + if (sub->next <= service_map.epoch) { + auto m = new MServiceMap(service_map); + sub->session->con->send_message(m); + if (sub->onetime) { + mon.with_session_map([sub](MonSessionMap& session_map) { + session_map.remove_sub(sub); + }); + } else { + sub->next = service_map.epoch + 1; + } + } +} + +void MgrStatMonitor::check_subs() +{ + dout(10) << __func__ << dendl; + if (!service_map.epoch) { + return; + } + auto subs = mon.session_map.subs.find("servicemap"); + if (subs == mon.session_map.subs.end()) { + return; + } + auto p = subs->second->begin(); + while (!p.end()) { + auto sub = *p; + ++p; + check_sub(sub); + } +} |