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