ceph/src/mon/MgrStatMonitor.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "MgrStatMonitor.h"
   5 #include "mon/OSDMonitor.h"
   6 #include "mon/PGMap.h"
   7 #include "mon/PGMonitor.h"
   8 #include "messages/MGetPoolStats.h"
   9 #include "messages/MGetPoolStatsReply.h"
  10 #include "messages/MMonMgrReport.h"
  11 #include "messages/MStatfs.h"
  12 #include "messages/MStatfsReply.h"
  13 #include "messages/MServiceMap.h"
  14
  15 class MgrPGStatService : public MonPGStatService {
  16   PGMapDigest& digest;
  17 public:
  18   MgrPGStatService(PGMapDigest& d) : digest(d) {}
  19
  20   const pool_stat_t* get_pool_stat(int poolid) const override {
  21     auto i = digest.pg_pool_sum.find(poolid);
  22     if (i != digest.pg_pool_sum.end()) {
  23       return &i->second;
  24     }
  25     return nullptr;
  26   }
  27
  28   ceph_statfs get_statfs() const override {
  29     return digest.get_statfs();
  30   }
  31
  32   void print_summary(Formatter *f, ostream *out) const override {
  33     digest.print_summary(f, out);
  34   }
  35   void dump_info(Formatter *f) const override {
  36     digest.dump(f);
  37   }
  38   void dump_fs_stats(stringstream *ss,
  39                      Formatter *f,
  40                      bool verbose) const override {
  41     digest.dump_fs_stats(ss, f, verbose);
  42   }
  43   void dump_pool_stats(const OSDMap& osdm, stringstream *ss, Formatter *f,
  44                        bool verbose) const override {
  45     digest.dump_pool_stats_full(osdm, ss, f, verbose);
  46   }
  47 };
  48
  49
  50 #define dout_subsys ceph_subsys_mon
  51 #undef dout_prefix
  52 #define dout_prefix _prefix(_dout, mon)
  53 static ostream& _prefix(std::ostream *_dout, Monitor *mon) {
  54   return *_dout << "mon." << mon->name << "@" << mon->rank
  55                 << "(" << mon->get_state_name()
  56                 << ").mgrstat ";
  57 }
  58
  59 MgrStatMonitor::MgrStatMonitor(Monitor *mn, Paxos *p, const string& service_name)
  60   : PaxosService(mn, p, service_name),
  61     pgservice(new MgrPGStatService(digest))
  62 {
  63 }
  64
  65 MgrStatMonitor::~MgrStatMonitor() = default;
  66
  67 MonPGStatService *MgrStatMonitor::get_pg_stat_service()
  68 {
  69   return pgservice.get();
  70 }
  71
  72 void MgrStatMonitor::create_initial()
  73 {
  74   dout(10) << __func__ << dendl;
  75   version = 0;
  76   service_map.epoch = 1;
  77   ::encode(service_map, pending_service_map_bl, CEPH_FEATURES_ALL);
  78 }
  79
  80 void MgrStatMonitor::update_from_paxos(bool *need_bootstrap)
  81 {
  82   version = get_last_committed();
  83   dout(10) << " " << version << dendl;
  84   load_health();
  85   bufferlist bl;
  86   get_version(version, bl);
  87   if (version) {
  88     assert(bl.length());
  89     try {
  90       auto p = bl.begin();
  91       ::decode(digest, p);
  92       ::decode(service_map, p);
  93       dout(10) << __func__ << " v" << version
  94                << " service_map e" << service_map.epoch << dendl;
  95     }
  96     catch (buffer::error& e) {
  97       derr << "failed to decode mgrstat state; luminous dev version?" << dendl;
  98     }
  99   }
 100   check_subs();
 101   update_logger();
 102 }
 103
 104 void MgrStatMonitor::update_logger()
 105 {
 106   dout(20) << __func__ << dendl;
 107   if (mon->osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
 108     dout(20) << "yielding cluster perfcounter updates to pgmon" << dendl;
 109     return;
 110   }
 111
 112   mon->cluster_logger->set(l_cluster_osd_bytes, digest.osd_sum.kb * 1024ull);
 113   mon->cluster_logger->set(l_cluster_osd_bytes_used,
 114                            digest.osd_sum.kb_used * 1024ull);
 115   mon->cluster_logger->set(l_cluster_osd_bytes_avail,
 116                            digest.osd_sum.kb_avail * 1024ull);
 117
 118   mon->cluster_logger->set(l_cluster_num_pool, digest.pg_pool_sum.size());
 119   uint64_t num_pg = 0;
 120   for (auto i : digest.num_pg_by_pool) {
 121     num_pg += i.second;
 122   }
 123   mon->cluster_logger->set(l_cluster_num_pg, num_pg);
 124
 125   unsigned active = 0, active_clean = 0, peering = 0;
 126   for (auto p = digest.num_pg_by_state.begin();
 127        p != digest.num_pg_by_state.end();
 128        ++p) {
 129     if (p->first & PG_STATE_ACTIVE) {
 130       active += p->second;
 131       if (p->first & PG_STATE_CLEAN)
 132         active_clean += p->second;
 133     }
 134     if (p->first & PG_STATE_PEERING)
 135       peering += p->second;
 136   }
 137   mon->cluster_logger->set(l_cluster_num_pg_active_clean, active_clean);
 138   mon->cluster_logger->set(l_cluster_num_pg_active, active);
 139   mon->cluster_logger->set(l_cluster_num_pg_peering, peering);
 140
 141   mon->cluster_logger->set(l_cluster_num_object, digest.pg_sum.stats.sum.num_objects);
 142   mon->cluster_logger->set(l_cluster_num_object_degraded, digest.pg_sum.stats.sum.num_objects_degraded);
 143   mon->cluster_logger->set(l_cluster_num_object_misplaced, digest.pg_sum.stats.sum.num_objects_misplaced);
 144   mon->cluster_logger->set(l_cluster_num_object_unfound, digest.pg_sum.stats.sum.num_objects_unfound);
 145   mon->cluster_logger->set(l_cluster_num_bytes, digest.pg_sum.stats.sum.num_bytes);
 146
 147 }
 148
 149 void MgrStatMonitor::create_pending()
 150 {
 151   dout(10) << " " << version << dendl;
 152   pending_digest = digest;
 153   pending_health_checks = get_health_checks();
 154   pending_service_map_bl.clear();
 155   ::encode(service_map, pending_service_map_bl, mon->get_quorum_con_features());
 156 }
 157
 158 void MgrStatMonitor::encode_pending(MonitorDBStore::TransactionRef t)
 159 {
 160   ++version;
 161   if (version < mon->pgmon()->get_last_committed()) {
 162     // fast-forward to pgmon version to ensure clients don't see a
 163     // jump back in time for MGetPoolStats and MStatFs.
 164     version = mon->pgmon()->get_last_committed() + 1;
 165   }
 166   dout(10) << " " << version << dendl;
 167   bufferlist bl;
 168   ::encode(pending_digest, bl, mon->get_quorum_con_features());
 169   assert(pending_service_map_bl.length());
 170   bl.append(pending_service_map_bl);
 171   put_version(t, version, bl);
 172   put_last_committed(t, version);
 173
 174   encode_health(pending_health_checks, t);
 175 }
 176
 177 version_t MgrStatMonitor::get_trim_to()
 178 {
 179   // we don't actually need *any* old states, but keep a few.
 180   if (version > 5) {
 181     return version - 5;
 182   }
 183   return 0;
 184 }
 185
 186 void MgrStatMonitor::on_active()
 187 {
 188   update_logger();
 189 }
 190
 191 void MgrStatMonitor::get_health(list<pair<health_status_t,string> >& summary,
 192                                 list<pair<health_status_t,string> > *detail,
 193                                 CephContext *cct) const
 194 {
 195 }
 196
 197 void MgrStatMonitor::tick()
 198 {
 199 }
 200
 201 void MgrStatMonitor::print_summary(Formatter *f, std::ostream *ss) const
 202 {
 203   pgservice->print_summary(f, ss);
 204 }
 205
 206 bool MgrStatMonitor::preprocess_query(MonOpRequestRef op)
 207 {
 208   auto m = static_cast<PaxosServiceMessage*>(op->get_req());
 209   switch (m->get_type()) {
 210   case CEPH_MSG_STATFS:
 211     return preprocess_statfs(op);
 212   case MSG_MON_MGR_REPORT:
 213     return preprocess_report(op);
 214   case MSG_GETPOOLSTATS:
 215     return preprocess_getpoolstats(op);
 216   default:
 217     mon->no_reply(op);
 218     derr << "Unhandled message type " << m->get_type() << dendl;
 219     return true;
 220   }
 221 }
 222
 223 bool MgrStatMonitor::prepare_update(MonOpRequestRef op)
 224 {
 225   auto m = static_cast<PaxosServiceMessage*>(op->get_req());
 226   switch (m->get_type()) {
 227   case MSG_MON_MGR_REPORT:
 228     return prepare_report(op);
 229   default:
 230     mon->no_reply(op);
 231     derr << "Unhandled message type " << m->get_type() << dendl;
 232     return true;
 233   }
 234 }
 235
 236 bool MgrStatMonitor::preprocess_report(MonOpRequestRef op)
 237 {
 238   return false;
 239 }
 240
 241 bool MgrStatMonitor::prepare_report(MonOpRequestRef op)
 242 {
 243   auto m = static_cast<MMonMgrReport*>(op->get_req());
 244   bufferlist bl = m->get_data();
 245   auto p = bl.begin();
 246   ::decode(pending_digest, p);
 247   pending_health_checks.swap(m->health_checks);
 248   if (m->service_map_bl.length()) {
 249     pending_service_map_bl.swap(m->service_map_bl);
 250   }
 251   dout(10) << __func__ << " " << pending_digest << ", "
 252            << pending_health_checks.checks.size() << " health checks" << dendl;
 253   return true;
 254 }
 255
 256 bool MgrStatMonitor::preprocess_getpoolstats(MonOpRequestRef op)
 257 {
 258   op->mark_pgmon_event(__func__);
 259   auto m = static_cast<MGetPoolStats*>(op->get_req());
 260   auto session = m->get_session();
 261   if (!session)
 262     return true;
 263   if (!session->is_capable("pg", MON_CAP_R)) {
 264     dout(0) << "MGetPoolStats received from entity with insufficient caps "
 265             << session->caps << dendl;
 266     return true;
 267   }
 268   if (m->fsid != mon->monmap->fsid) {
 269     dout(0) << __func__ << " on fsid "
 270             << m->fsid << " != " << mon->monmap->fsid << dendl;
 271     return true;
 272   }
 273   epoch_t ver = 0;
 274   if (mon->pgservice == get_pg_stat_service()) {
 275     ver = get_last_committed();
 276   } else {
 277     ver = mon->pgmon()->get_last_committed();
 278   }
 279   auto reply = new MGetPoolStatsReply(m->fsid, m->get_tid(), ver);
 280   for (const auto& pool_name : m->pools) {
 281     const auto pool_id = mon->osdmon()->osdmap.lookup_pg_pool_name(pool_name);
 282     if (pool_id == -ENOENT)
 283       continue;
 284     auto pool_stat = mon->pgservice->get_pool_stat(pool_id);
 285     if (!pool_stat)
 286       continue;
 287     reply->pool_stats[pool_name] = *pool_stat;
 288   }
 289   mon->send_reply(op, reply);
 290   return true;
 291 }
 292
 293 bool MgrStatMonitor::preprocess_statfs(MonOpRequestRef op)
 294 {
 295   op->mark_pgmon_event(__func__);
 296   auto statfs = static_cast<MStatfs*>(op->get_req());
 297   auto session = statfs->get_session();
 298   if (!session)
 299     return true;
 300   if (!session->is_capable("pg", MON_CAP_R)) {
 301     dout(0) << "MStatfs received from entity with insufficient privileges "
 302             << session->caps << dendl;
 303     return true;
 304   }
 305   if (statfs->fsid != mon->monmap->fsid) {
 306     dout(0) << __func__ << " on fsid " << statfs->fsid
 307             << " != " << mon->monmap->fsid << dendl;
 308     return true;
 309   }
 310   dout(10) << __func__ << " " << *statfs
 311            << " from " << statfs->get_orig_source() << dendl;
 312   epoch_t ver = 0;
 313   if (mon->pgservice == get_pg_stat_service()) {
 314     ver = get_last_committed();
 315   } else {
 316     ver = mon->pgmon()->get_last_committed();
 317   }
 318   auto reply = new MStatfsReply(statfs->fsid, statfs->get_tid(), ver);
 319   reply->h.st = mon->pgservice->get_statfs();
 320   mon->send_reply(op, reply);
 321   return true;
 322 }
 323
 324 void MgrStatMonitor::check_sub(Subscription *sub)
 325 {
 326   const auto epoch = mon->monmap->get_epoch();
 327   dout(10) << __func__
 328            << " next " << sub->next
 329            << " have " << epoch << dendl;
 330   if (sub->next <= service_map.epoch) {
 331     auto m = new MServiceMap(service_map);
 332     sub->session->con->send_message(m);
 333     if (sub->onetime) {
 334       mon->with_session_map([this, sub](MonSessionMap& session_map) {
 335           session_map.remove_sub(sub);
 336         });
 337     } else {
 338       sub->next = epoch + 1;
 339     }
 340   }
 341 }
 342
 343 void MgrStatMonitor::check_subs()
 344 {
 345   dout(10) << __func__ << dendl;
 346   if (!service_map.epoch) {
 347     return;
 348   }
 349   auto subs = mon->session_map.subs.find("servicemap");
 350   if (subs == mon->session_map.subs.end()) {
 351     return;
 352   }
 353   auto p = subs->second->begin();
 354   while (!p.end()) {
 355     auto sub = *p;
 356     ++p;
 357     check_sub(sub);
 358   }
 359 }