[ceph.git] / ceph / src / mon / MgrStatMonitor.cc

// -*- 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/PGMap.h"
#include "mon/PGMonitor.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"

class MgrPGStatService : public MonPGStatService {
  PGMapDigest& digest;
public:
  MgrPGStatService(PGMapDigest& d) : digest(d) {}

  const pool_stat_t* get_pool_stat(int64_t poolid) const override {
    auto i = digest.pg_pool_sum.find(poolid);
    if (i != digest.pg_pool_sum.end()) {
      return &i->second;
    }
    return nullptr;
  }

  ceph_statfs get_statfs(OSDMap& osdmap,
			 boost::optional<int64_t> data_pool) const override {
    return digest.get_statfs(osdmap, data_pool);
  }

  void print_summary(Formatter *f, ostream *out) const override {
    digest.print_summary(f, out);
  }
  void dump_info(Formatter *f) const override {
    digest.dump(f);
  }
  void dump_fs_stats(stringstream *ss,
		     Formatter *f,
		     bool verbose) const override {
    digest.dump_fs_stats(ss, f, verbose);
  }
  void dump_pool_stats(const OSDMap& osdm, stringstream *ss, Formatter *f,
		       bool verbose) const override {
    digest.dump_pool_stats_full(osdm, ss, f, verbose);
  }
};


#define dout_subsys ceph_subsys_mon
#undef dout_prefix
#define dout_prefix _prefix(_dout, mon)
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),
    pgservice(new MgrPGStatService(digest))
{
}

MgrStatMonitor::~MgrStatMonitor() = default;

MonPGStatService *MgrStatMonitor::get_pg_stat_service()
{
  return pgservice.get();
}

void MgrStatMonitor::create_initial()
{
  dout(10) << __func__ << dendl;
  version = 0;
  service_map.epoch = 1;
  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) {
    assert(bl.length());
    try {
      auto p = bl.begin();
      ::decode(digest, p);
      ::decode(service_map, p);
      dout(10) << __func__ << " v" << version
	       << " service_map e" << service_map.epoch << dendl;
    }
    catch (buffer::error& e) {
      derr << "failed to decode mgrstat state; luminous dev version? "
	   << e.what() << dendl;
    }
  }
  check_subs();
  update_logger();
}

void MgrStatMonitor::update_logger()
{
  dout(20) << __func__ << dendl;
  if (mon->osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
    dout(20) << "yielding cluster perfcounter updates to pgmon" << dendl;
    return;
  }

  mon->cluster_logger->set(l_cluster_osd_bytes, digest.osd_sum.kb * 1024ull);
  mon->cluster_logger->set(l_cluster_osd_bytes_used,
                           digest.osd_sum.kb_used * 1024ull);
  mon->cluster_logger->set(l_cluster_osd_bytes_avail,
                           digest.osd_sum.kb_avail * 1024ull);

  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;
  if (version < mon->pgmon()->get_last_committed()) {
    // fast-forward to pgmon version to ensure clients don't see a
    // jump back in time for MGetPoolStats and MStatFs.
    version = mon->pgmon()->get_last_committed() + 1;
  }
  dout(10) << " " << version << dendl;
  bufferlist bl;
  ::encode(pending_digest, bl, mon->get_quorum_con_features());
  assert(pending_service_map_bl.length());
  bl.append(pending_service_map_bl);
  put_version(t, version, bl);
  put_last_committed(t, version);

  encode_health(pending_health_checks, t);
}

version_t MgrStatMonitor::get_trim_to()
{
  // 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::get_health(list<pair<health_status_t,string> >& summary,
				list<pair<health_status_t,string> > *detail,
				CephContext *cct) const
{
}

void MgrStatMonitor::tick()
{
}

void MgrStatMonitor::print_summary(Formatter *f, std::ostream *ss) const
{
  pgservice->print_summary(f, ss);
}

bool MgrStatMonitor::preprocess_query(MonOpRequestRef op)
{
  auto m = static_cast<PaxosServiceMessage*>(op->get_req());
  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 = static_cast<PaxosServiceMessage*>(op->get_req());
  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)
{
  mon->no_reply(op);
  return false;
}

bool MgrStatMonitor::prepare_report(MonOpRequestRef op)
{
  auto m = static_cast<MMonMgrReport*>(op->get_req());
  bufferlist bl = m->get_data();
  auto p = bl.begin();
  ::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);
  }
  dout(10) << __func__ << " " << pending_digest << ", "
	   << pending_health_checks.checks.size() << " health checks" << dendl;
  return true;
}

bool MgrStatMonitor::preprocess_getpoolstats(MonOpRequestRef op)
{
  op->mark_pgmon_event(__func__);
  auto m = static_cast<MGetPoolStats*>(op->get_req());
  auto session = m->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 = 0;
  if (mon->pgservice == get_pg_stat_service()) {
    ver = get_last_committed();
  } else {
    ver = mon->pgmon()->get_last_committed();
  }
  auto reply = new MGetPoolStatsReply(m->fsid, m->get_tid(), ver);
  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 = mon->pgservice->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 = static_cast<MStatfs*>(op->get_req());
  auto session = statfs->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;
  }
  dout(10) << __func__ << " " << *statfs
           << " from " << statfs->get_orig_source() << dendl;
  epoch_t ver = 0;
  if (mon->pgservice == get_pg_stat_service()) {
    ver = get_last_committed();
  } else {
    ver = mon->pgmon()->get_last_committed();
  }
  auto reply = new MStatfsReply(statfs->fsid, statfs->get_tid(), ver);
  reply->h.st = mon->pgservice->get_statfs(mon->osdmon()->osdmap,
					   statfs->data_pool);
  mon->send_reply(op, reply);
  return true;
}

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