+++ /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) 2004-2006 Sage Weil <sage@newdream.net>
- *
- * 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 "json_spirit/json_spirit.h"
-#include "common/debug.h" // undo damage
-#include "PGMonitor.h"
-#include "Monitor.h"
-#include "OSDMonitor.h"
-#include "MonitorDBStore.h"
-#include "PGStatService.h"
-
-#include "messages/MPGStats.h"
-#include "messages/MPGStatsAck.h"
-
-#include "messages/MOSDPGCreate.h"
-#include "messages/MMonCommand.h"
-#include "messages/MOSDScrub.h"
-
-#include "common/Formatter.h"
-#include "common/config.h"
-
-#include "include/stringify.h"
-
-#include "osd/osd_types.h"
-
-#include "common/config.h"
-#include "common/errno.h"
-#include "common/strtol.h"
-#include "include/str_list.h"
-#include <sstream>
-
-#define dout_subsys ceph_subsys_mon
-#undef dout_prefix
-#define dout_prefix _prefix(_dout, mon, pg_map)
-static ostream& _prefix(std::ostream *_dout, const Monitor *mon, const PGMap& pg_map) {
- return *_dout << "mon." << mon->name << "@" << mon->rank
- << "(" << mon->get_state_name()
- << ").pg v" << pg_map.version << " ";
-}
-
-/*
- Tick function to update the map based on performance every N seconds
- */
-
-void PGMonitor::on_restart()
-{
- // clear leader state
- last_osd_report.clear();
-}
-
-void PGMonitor::on_active()
-{
- if (mon->is_leader()) {
- check_all_pgs = true;
- check_osd_map(mon->osdmon()->osdmap.get_epoch());
- }
-
- update_logger();
-
- if (mon->is_leader() &&
- mon->osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
- mon->clog->info() << "pgmap " << pg_map;
- }
-}
-
-void PGMonitor::update_logger()
-{
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- return;
- }
- dout(10) << "update_logger" << dendl;
-
- mon->cluster_logger->set(l_cluster_osd_bytes, pg_map.osd_sum.kb * 1024ull);
- mon->cluster_logger->set(l_cluster_osd_bytes_used,
- pg_map.osd_sum.kb_used * 1024ull);
- mon->cluster_logger->set(l_cluster_osd_bytes_avail,
- pg_map.osd_sum.kb_avail * 1024ull);
-
- mon->cluster_logger->set(l_cluster_num_pool, pg_map.pg_pool_sum.size());
- mon->cluster_logger->set(l_cluster_num_pg, pg_map.pg_stat.size());
-
- unsigned active = 0, active_clean = 0, peering = 0;
- for (ceph::unordered_map<int,int>::iterator p = pg_map.num_pg_by_state.begin();
- p != pg_map.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, pg_map.pg_sum.stats.sum.num_objects);
- mon->cluster_logger->set(l_cluster_num_object_degraded, pg_map.pg_sum.stats.sum.num_objects_degraded);
- mon->cluster_logger->set(l_cluster_num_object_misplaced, pg_map.pg_sum.stats.sum.num_objects_misplaced);
- mon->cluster_logger->set(l_cluster_num_object_unfound, pg_map.pg_sum.stats.sum.num_objects_unfound);
- mon->cluster_logger->set(l_cluster_num_bytes, pg_map.pg_sum.stats.sum.num_bytes);
-}
-
-void PGMonitor::tick()
-{
- if (!is_active()) return;
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- return;
- }
-
- handle_osd_timeouts();
-
- if (!pg_map.pg_sum_deltas.empty()) {
- utime_t age = ceph_clock_now() - pg_map.stamp;
- if (age > 2 * g_conf->mon_delta_reset_interval) {
- dout(10) << " clearing pg_map delta (" << age << " > " << g_conf->mon_delta_reset_interval << " seconds old)" << dendl;
- pg_map.clear_delta();
- }
- }
-
- /* If we have deltas for pools, run through pgmap's 'per_pool_sum_delta' and
- * clear any deltas that are old enough.
- *
- * Note that 'per_pool_sum_delta' keeps a pool id as key, and a pair containing
- * the calc'ed stats delta and an absolute timestamp from when those stats were
- * obtained -- the timestamp IS NOT a delta itself.
- */
- if (!pg_map.per_pool_sum_deltas.empty()) {
- ceph::unordered_map<uint64_t,pair<pool_stat_t,utime_t> >::iterator it;
- for (it = pg_map.per_pool_sum_delta.begin();
- it != pg_map.per_pool_sum_delta.end(); ) {
- utime_t age = ceph_clock_now() - it->second.second;
- if (age > 2*g_conf->mon_delta_reset_interval) {
- dout(10) << " clearing pg_map delta for pool " << it->first
- << " (" << age << " > " << g_conf->mon_delta_reset_interval
- << " seconds old)" << dendl;
- pg_map.per_pool_sum_deltas.erase(it->first);
- pg_map.per_pool_sum_deltas_stamps.erase(it->first);
- pg_map.per_pool_sum_delta.erase((it++)->first);
- } else {
- ++it;
- }
- }
- }
-
- dout(10) << pg_map << dendl;
-}
-
-void PGMonitor::create_initial()
-{
- dout(10) << "create_initial -- creating initial map" << dendl;
- format_version = 1;
-}
-
-void PGMonitor::update_from_paxos(bool *need_bootstrap)
-{
- if (did_delete)
- return;
-
- if (get_value("deleted")) {
- did_delete = true;
- dout(10) << __func__ << " deleted, clearing in-memory PGMap" << dendl;
- pg_map = PGMap();
- pending_inc = PGMap::Incremental();
- pgservice.reset();
- last_osd_report.clear();
- return;
- }
-
- version_t version = get_last_committed();
- if (version == pg_map.version)
- return;
-
- assert(version >= pg_map.version);
- if (format_version < 1) {
- derr << __func__ << "unsupported monitor protocol: "
- << get_service_name() << ".format_version = "
- << format_version << dendl;
- }
- assert(format_version >= 1);
-
- // pg/osd keys in leveldb
- // read meta
- while (version > pg_map.version) {
- // load full state?
- if (pg_map.version == 0) {
- dout(10) << __func__ << " v0, read_full" << dendl;
- read_pgmap_full();
- goto out;
- }
-
- // incremental state?
- dout(10) << __func__ << " read_incremental" << dendl;
- bufferlist bl;
- int r = get_version(pg_map.version + 1, bl);
- if (r == -ENOENT) {
- dout(10) << __func__ << " failed to read_incremental, read_full" << dendl;
- // reset pg map
- pg_map = PGMap();
- read_pgmap_full();
- goto out;
- }
- assert(r == 0);
- apply_pgmap_delta(bl);
- }
-
- read_pgmap_meta();
-
-out:
- assert(version == pg_map.version);
-
- update_logger();
-}
-
-void PGMonitor::on_upgrade()
-{
- dout(1) << __func__ << " discarding in-core PGMap" << dendl;
- pg_map = PGMap();
-}
-
-void PGMonitor::upgrade_format()
-{
- unsigned current = 1;
- assert(format_version == current);
-}
-
-void PGMonitor::post_paxos_update()
-{
- if (did_delete)
- return;
- dout(10) << __func__ << dendl;
- OSDMap& osdmap = mon->osdmon()->osdmap;
- if (mon->monmap->get_required_features().contains_all(
- ceph::features::mon::FEATURE_LUMINOUS)) {
- // let OSDMonitor take care of the pg-creates subscriptions.
- return;
- }
- if (osdmap.get_epoch()) {
- if (osdmap.get_num_up_osds() > 0) {
- assert(osdmap.get_up_osd_features() & CEPH_FEATURE_MON_STATEFUL_SUB);
- check_subs();
- }
- }
-}
-
-void PGMonitor::handle_osd_timeouts()
-{
- if (!mon->is_leader())
- return;
- if (did_delete)
- return;
-
- utime_t now(ceph_clock_now());
- utime_t timeo(g_conf->mon_osd_report_timeout, 0);
- if (now - mon->get_leader_since() < timeo) {
- // We haven't been the leader for long enough to consider OSD timeouts
- return;
- }
-
- if (mon->osdmon()->is_writeable())
- mon->osdmon()->handle_osd_timeouts(now, last_osd_report);
-}
-
-void PGMonitor::create_pending()
-{
- if (did_delete)
- return;
- do_delete = false;
- pending_inc = PGMap::Incremental();
- pending_inc.version = pg_map.version + 1;
- if (pg_map.version == 0) {
- // pull initial values from first leader mon's config
- pending_inc.full_ratio = g_conf->mon_osd_full_ratio;
- if (pending_inc.full_ratio > 1.0)
- pending_inc.full_ratio /= 100.0;
- pending_inc.nearfull_ratio = g_conf->mon_osd_nearfull_ratio;
- if (pending_inc.nearfull_ratio > 1.0)
- pending_inc.nearfull_ratio /= 100.0;
- } else {
- pending_inc.full_ratio = pg_map.full_ratio;
- pending_inc.nearfull_ratio = pg_map.nearfull_ratio;
- }
- dout(10) << "create_pending v " << pending_inc.version << dendl;
-}
-
-void PGMonitor::read_pgmap_meta()
-{
- dout(10) << __func__ << dendl;
-
- string prefix = pgmap_meta_prefix;
-
- version_t version = mon->store->get(prefix, "version");
- epoch_t last_osdmap_epoch = mon->store->get(prefix, "last_osdmap_epoch");
- epoch_t last_pg_scan = mon->store->get(prefix, "last_pg_scan");
- pg_map.set_version(version);
- pg_map.set_last_osdmap_epoch(last_osdmap_epoch);
-
- if (last_pg_scan != pg_map.get_last_pg_scan()) {
- pg_map.set_last_pg_scan(last_pg_scan);
- }
-
- float full_ratio, nearfull_ratio;
- {
- bufferlist bl;
- mon->store->get(prefix, "full_ratio", bl);
- bufferlist::iterator p = bl.begin();
- ::decode(full_ratio, p);
- }
- {
- bufferlist bl;
- mon->store->get(prefix, "nearfull_ratio", bl);
- bufferlist::iterator p = bl.begin();
- ::decode(nearfull_ratio, p);
- }
- pg_map.set_full_ratios(full_ratio, nearfull_ratio);
- {
- bufferlist bl;
- mon->store->get(prefix, "stamp", bl);
- bufferlist::iterator p = bl.begin();
- utime_t stamp;
- ::decode(stamp, p);
- pg_map.set_stamp(stamp);
- }
-}
-
-void PGMonitor::read_pgmap_full()
-{
- read_pgmap_meta();
-
- string prefix = pgmap_pg_prefix;
- for (KeyValueDB::Iterator i = mon->store->get_iterator(prefix); i->valid(); i->next()) {
- string key = i->key();
- pg_t pgid;
- if (!pgid.parse(key.c_str())) {
- dout(0) << "unable to parse key " << key << dendl;
- continue;
- }
- bufferlist bl = i->value();
- pg_map.update_pg(pgid, bl);
- dout(20) << " got " << pgid << dendl;
- }
-
- prefix = pgmap_osd_prefix;
- for (KeyValueDB::Iterator i = mon->store->get_iterator(prefix); i->valid(); i->next()) {
- string key = i->key();
- int osd = atoi(key.c_str());
- bufferlist bl = i->value();
- pg_map.update_osd(osd, bl);
- dout(20) << " got osd." << osd << dendl;
- }
-}
-
-void PGMonitor::apply_pgmap_delta(bufferlist& bl)
-{
- version_t v = pg_map.version + 1;
-
- utime_t inc_stamp;
- bufferlist dirty_pgs, dirty_osds;
- {
- bufferlist::iterator p = bl.begin();
- ::decode(inc_stamp, p);
- ::decode(dirty_pgs, p);
- ::decode(dirty_osds, p);
- }
-
- pool_stat_t pg_sum_old = pg_map.pg_sum;
- mempool::pgmap::unordered_map<uint64_t, pool_stat_t> pg_pool_sum_old;
-
- // pgs
- set<int64_t> deleted_pools;
- bufferlist::iterator p = dirty_pgs.begin();
- while (!p.end()) {
- pg_t pgid;
- ::decode(pgid, p);
-
- int r;
- bufferlist pgbl;
- if (deleted_pools.count(pgid.pool())) {
- r = -ENOENT;
- } else {
- r = mon->store->get(pgmap_pg_prefix, stringify(pgid), pgbl);
- if (pg_pool_sum_old.count(pgid.pool()) == 0)
- pg_pool_sum_old[pgid.pool()] = pg_map.pg_pool_sum[pgid.pool()];
- }
-
- if (r >= 0) {
- pg_map.update_pg(pgid, pgbl);
- dout(20) << " refreshing pg " << pgid
- << " " << pg_map.pg_stat[pgid].reported_epoch
- << ":" << pg_map.pg_stat[pgid].reported_seq
- << " " << pg_state_string(pg_map.pg_stat[pgid].state)
- << dendl;
- } else {
- dout(20) << " removing pg " << pgid << dendl;
- pg_map.remove_pg(pgid);
- if (pgid.ps() == 0)
- deleted_pools.insert(pgid.pool());
- }
- }
-
- // osds
- p = dirty_osds.begin();
- while (!p.end()) {
- int32_t osd;
- ::decode(osd, p);
- dout(20) << " refreshing osd." << osd << dendl;
- bufferlist bl;
- int r = mon->store->get(pgmap_osd_prefix, stringify(osd), bl);
- if (r >= 0) {
- pg_map.update_osd(osd, bl);
- } else {
- pg_map.remove_osd(osd);
- }
- }
-
- pg_map.update_global_delta(g_ceph_context, inc_stamp, pg_sum_old);
- pg_map.update_pool_deltas(g_ceph_context, inc_stamp, pg_pool_sum_old);
-
- // clean up deleted pools after updating the deltas
- for (set<int64_t>::iterator p = deleted_pools.begin();
- p != deleted_pools.end();
- ++p) {
- dout(20) << " deleted pool " << *p << dendl;
- pg_map.deleted_pool(*p);
- }
-
- // ok, we're now on the new version
- pg_map.version = v;
-}
-
-
-void PGMonitor::encode_pending(MonitorDBStore::TransactionRef t)
-{
- if (did_delete)
- return;
-
- string prefix = pgmap_meta_prefix;
- if (do_delete) {
- dout(1) << __func__ << " clearing pgmap data at v" << pending_inc.version
- << dendl;
- do_delete = false;
- for (auto key : { "version", "stamp", "last_osdmap_epoch",
- "last_pg_scan", "full_ratio", "nearfull_ratio" }) {
- t->erase(prefix, key);
- }
- for (auto& p : pg_map.pg_stat) {
- t->erase(prefix, stringify(p.first));
- }
- for (auto& p : pg_map.osd_stat) {
- t->erase(prefix, stringify(p.first));
- }
- put_last_committed(t, pending_inc.version);
- put_value(t, "deleted", 1);
- return;
- }
-
- assert(mon->osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS ||
- pending_inc.version == 1 /* rebuild-mondb.yaml case */);
-
- version_t version = pending_inc.version;
- dout(10) << __func__ << " v " << version << dendl;
- assert(get_last_committed() + 1 == version);
- pending_inc.stamp = ceph_clock_now();
-
- uint64_t features = mon->get_quorum_con_features();
-
- t->put(prefix, "version", pending_inc.version);
- {
- bufferlist bl;
- ::encode(pending_inc.stamp, bl);
- t->put(prefix, "stamp", bl);
- }
-
- if (pending_inc.osdmap_epoch)
- t->put(prefix, "last_osdmap_epoch", pending_inc.osdmap_epoch);
- if (pending_inc.pg_scan)
- t->put(prefix, "last_pg_scan", pending_inc.pg_scan);
- if (pending_inc.full_ratio > 0) {
- bufferlist bl;
- ::encode(pending_inc.full_ratio, bl);
- t->put(prefix, "full_ratio", bl);
- }
- if (pending_inc.nearfull_ratio > 0) {
- bufferlist bl;
- ::encode(pending_inc.nearfull_ratio, bl);
- t->put(prefix, "nearfull_ratio", bl);
- }
-
- bufferlist incbl;
- ::encode(pending_inc.stamp, incbl);
- {
- bufferlist dirty;
- string prefix = pgmap_pg_prefix;
- for (map<pg_t,pg_stat_t>::const_iterator p = pending_inc.pg_stat_updates.begin();
- p != pending_inc.pg_stat_updates.end();
- ++p) {
- ::encode(p->first, dirty);
- bufferlist bl;
- ::encode(p->second, bl, features);
- t->put(prefix, stringify(p->first), bl);
- }
- for (set<pg_t>::const_iterator p = pending_inc.pg_remove.begin(); p != pending_inc.pg_remove.end(); ++p) {
- ::encode(*p, dirty);
- t->erase(prefix, stringify(*p));
- }
- ::encode(dirty, incbl);
- }
- {
- bufferlist dirty;
- string prefix = pgmap_osd_prefix;
- for (map<int32_t,osd_stat_t>::const_iterator p =
- pending_inc.get_osd_stat_updates().begin();
- p != pending_inc.get_osd_stat_updates().end();
- ++p) {
- ::encode(p->first, dirty);
- bufferlist bl;
- ::encode(p->second, bl, features);
- ::encode(pending_inc.get_osd_epochs().find(p->first)->second, bl);
- t->put(prefix, stringify(p->first), bl);
- }
- for (set<int32_t>::const_iterator p =
- pending_inc.get_osd_stat_rm().begin();
- p != pending_inc.get_osd_stat_rm().end();
- ++p) {
- ::encode(*p, dirty);
- t->erase(prefix, stringify(*p));
- }
- ::encode(dirty, incbl);
- }
-
- put_version(t, version, incbl);
-
- put_last_committed(t, version);
-}
-
-version_t PGMonitor::get_trim_to()
-{
- unsigned max = g_conf->mon_max_pgmap_epochs;
- version_t version = get_last_committed();
- if (mon->is_leader() && (version > max))
- return version - max;
-
- return 0;
-}
-
-bool PGMonitor::preprocess_query(MonOpRequestRef op)
-{
- op->mark_pgmon_event(__func__);
- PaxosServiceMessage *m = static_cast<PaxosServiceMessage*>(op->get_req());
- dout(10) << "preprocess_query " << *m
- << " from " << m->get_orig_source_inst() << dendl;
- switch (m->get_type()) {
- case MSG_PGSTATS:
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- return true;
- }
- return preprocess_pg_stats(op);
-
- case MSG_MON_COMMAND:
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- bufferlist rdata;
- mon->reply_command(op, -EOPNOTSUPP, "this command is obsolete", rdata,
- get_last_committed());
- return true;
- }
- return preprocess_command(op);
-
- default:
- ceph_abort();
- return true;
- }
-}
-
-bool PGMonitor::prepare_update(MonOpRequestRef op)
-{
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- return false;
- }
-
- op->mark_pgmon_event(__func__);
- PaxosServiceMessage *m = static_cast<PaxosServiceMessage*>(op->get_req());
- dout(10) << "prepare_update " << *m << " from " << m->get_orig_source_inst() << dendl;
- switch (m->get_type()) {
- case MSG_PGSTATS:
- return prepare_pg_stats(op);
-
- case MSG_MON_COMMAND:
- return prepare_command(op);
-
- default:
- ceph_abort();
- return false;
- }
-}
-
-bool PGMonitor::preprocess_pg_stats(MonOpRequestRef op)
-{
- op->mark_pgmon_event(__func__);
- MPGStats *stats = static_cast<MPGStats*>(op->get_req());
- // check caps
- MonSession *session = stats->get_session();
- if (!session) {
- dout(10) << "PGMonitor::preprocess_pg_stats: no monitor session!" << dendl;
- return true;
- }
- if (!session->is_capable("pg", MON_CAP_R)) {
- derr << "PGMonitor::preprocess_pg_stats: MPGStats received from entity "
- << "with insufficient privileges " << session->caps << dendl;
- return true;
- }
-
- if (stats->fsid != mon->monmap->fsid) {
- dout(0) << __func__ << " drop message on fsid " << stats->fsid << " != "
- << mon->monmap->fsid << " for " << *stats << dendl;
- return true;
- }
-
- // First, just see if they need a new osdmap. But
- // only if they've had the map for a while.
- if (stats->had_map_for > 30.0 &&
- mon->osdmon()->is_readable() &&
- stats->epoch < mon->osdmon()->osdmap.get_epoch() &&
- !session->proxy_con)
- mon->osdmon()->send_latest_now_nodelete(op, stats->epoch+1);
-
- // Always forward the PGStats to the leader, even if they are the same as
- // the old PGStats. The leader will mark as down osds that haven't sent
- // PGStats for a few minutes.
- return false;
-}
-
-bool PGMonitor::pg_stats_have_changed(int from, const MPGStats *stats) const
-{
- // any new osd info?
- ceph::unordered_map<int,osd_stat_t>::const_iterator s = pg_map.osd_stat.find(from);
- if (s == pg_map.osd_stat.end())
- return true;
-
- if (s->second != stats->osd_stat)
- return true;
-
- // any new pg info?
- for (map<pg_t,pg_stat_t>::const_iterator p = stats->pg_stat.begin();
- p != stats->pg_stat.end(); ++p) {
- ceph::unordered_map<pg_t,pg_stat_t>::const_iterator t = pg_map.pg_stat.find(p->first);
- if (t == pg_map.pg_stat.end())
- return true;
-
- if (t->second.reported_epoch != p->second.reported_epoch ||
- t->second.reported_seq != p->second.reported_seq)
- return true;
- }
-
- return false;
-}
-
-struct PGMonitor::C_Stats : public C_MonOp {
- PGMonitor *pgmon;
- MonOpRequestRef stats_op_ack;
- entity_inst_t who;
- C_Stats(PGMonitor *p,
- MonOpRequestRef op,
- MonOpRequestRef op_ack)
- : C_MonOp(op), pgmon(p), stats_op_ack(op_ack) {}
- void _finish(int r) override {
- if (r >= 0) {
- pgmon->_updated_stats(op, stats_op_ack);
- } else if (r == -ECANCELED) {
- return;
- } else if (r == -EAGAIN) {
- pgmon->dispatch(op);
- } else {
- assert(0 == "bad C_Stats return value");
- }
- }
-};
-
-bool PGMonitor::prepare_pg_stats(MonOpRequestRef op)
-{
- op->mark_pgmon_event(__func__);
- MPGStats *stats = static_cast<MPGStats*>(op->get_req());
- dout(10) << "prepare_pg_stats " << *stats << " from " << stats->get_orig_source() << dendl;
- int from = stats->get_orig_source().num();
-
- if (stats->fsid != mon->monmap->fsid) {
- dout(0) << "prepare_pg_stats on fsid " << stats->fsid << " != " << mon->monmap->fsid << dendl;
- return false;
- }
-
- if (!stats->get_orig_source().is_osd() ||
- !mon->osdmon()->osdmap.is_up(from) ||
- stats->get_orig_source_inst() != mon->osdmon()->osdmap.get_inst(from)) {
- dout(1) << " ignoring stats from non-active osd." << dendl;
- return false;
- }
-
- last_osd_report[from] = ceph_clock_now();
-
- if (!pg_stats_have_changed(from, stats)) {
- dout(10) << " message contains no new osd|pg stats" << dendl;
- MPGStatsAck *ack = new MPGStatsAck;
- ack->set_tid(stats->get_tid());
- for (map<pg_t,pg_stat_t>::const_iterator p = stats->pg_stat.begin();
- p != stats->pg_stat.end();
- ++p) {
- ack->pg_stat[p->first] = make_pair(p->second.reported_seq, p->second.reported_epoch);
- }
- mon->send_reply(op, ack);
- return false;
- }
-
- // osd stat
- if (mon->osdmon()->osdmap.is_in(from)) {
- pending_inc.update_stat(from, stats->epoch, std::move(stats->osd_stat));
- } else {
- pending_inc.update_stat(from, stats->epoch, osd_stat_t());
- }
-
- if (pg_map.osd_stat.count(from))
- dout(10) << " got osd." << from << " " << stats->osd_stat << " (was " << pg_map.osd_stat[from] << ")" << dendl;
- else
- dout(10) << " got osd." << from << " " << stats->osd_stat << " (first report)" << dendl;
-
- // pg stats
- MPGStatsAck *ack = new MPGStatsAck;
- MonOpRequestRef ack_op = mon->op_tracker.create_request<MonOpRequest>(ack);
- ack->set_tid(stats->get_tid());
- for (map<pg_t,pg_stat_t>::iterator p = stats->pg_stat.begin();
- p != stats->pg_stat.end();
- ++p) {
- pg_t pgid = p->first;
- ack->pg_stat[pgid] = make_pair(p->second.reported_seq, p->second.reported_epoch);
-
- if (pg_map.pg_stat.count(pgid) &&
- pg_map.pg_stat[pgid].get_version_pair() > p->second.get_version_pair()) {
- dout(15) << " had " << pgid << " from " << pg_map.pg_stat[pgid].reported_epoch << ":"
- << pg_map.pg_stat[pgid].reported_seq << dendl;
- continue;
- }
- if (pending_inc.pg_stat_updates.count(pgid) &&
- pending_inc.pg_stat_updates[pgid].get_version_pair() > p->second.get_version_pair()) {
- dout(15) << " had " << pgid << " from " << pending_inc.pg_stat_updates[pgid].reported_epoch << ":"
- << pending_inc.pg_stat_updates[pgid].reported_seq << " (pending)" << dendl;
- continue;
- }
-
- if (pg_map.pg_stat.count(pgid) == 0) {
- dout(15) << " got " << pgid << " reported at " << p->second.reported_epoch << ":"
- << p->second.reported_seq
- << " state " << pg_state_string(p->second.state)
- << " but DNE in pg_map; pool was probably deleted."
- << dendl;
- continue;
- }
-
- dout(15) << " got " << pgid
- << " reported at " << p->second.reported_epoch << ":" << p->second.reported_seq
- << " state " << pg_state_string(pg_map.pg_stat[pgid].state)
- << " -> " << pg_state_string(p->second.state)
- << dendl;
- pending_inc.pg_stat_updates[pgid] = p->second;
- }
-
- wait_for_finished_proposal(op, new C_Stats(this, op, ack_op));
- return true;
-}
-
-void PGMonitor::_updated_stats(MonOpRequestRef op, MonOpRequestRef ack_op)
-{
- op->mark_pgmon_event(__func__);
- ack_op->mark_pgmon_event(__func__);
- MPGStats *ack = static_cast<MPGStats*>(ack_op->get_req());
- ack->get(); // MonOpRequestRef owns one ref; give the other to send_reply.
- dout(7) << "_updated_stats for "
- << op->get_req()->get_orig_source_inst() << dendl;
- mon->send_reply(op, ack);
-}
-
-
-// ------------------------
-
-struct RetryCheckOSDMap : public Context {
- PGMonitor *pgmon;
- epoch_t epoch;
- RetryCheckOSDMap(PGMonitor *p, epoch_t e) : pgmon(p), epoch(e) {
- }
- void finish(int r) override {
- if (r == -ECANCELED)
- return;
-
- pgmon->check_osd_map(epoch);
- }
-};
-
-void PGMonitor::check_osd_map(epoch_t epoch)
-{
- if (mon->is_peon())
- return; // whatever.
-
- if (did_delete)
- return;
-
- if (pg_map.last_osdmap_epoch >= epoch) {
- dout(10) << __func__ << " already seen " << pg_map.last_osdmap_epoch
- << " >= " << epoch << dendl;
- return;
- }
-
- if (!mon->osdmon()->is_readable()) {
- dout(10) << __func__ << " -- osdmap not readable, waiting" << dendl;
- mon->osdmon()->wait_for_readable_ctx(new RetryCheckOSDMap(this, epoch));
- return;
- }
-
- if (!is_writeable()) {
- dout(10) << __func__ << " -- pgmap not writeable, waiting" << dendl;
- wait_for_writeable_ctx(new RetryCheckOSDMap(this, epoch));
- return;
- }
-
- const OSDMap& osdmap = mon->osdmon()->osdmap;
- if (!did_delete && osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- // delete all my data
- dout(1) << __func__ << " will clear pg_map data" << dendl;
- do_delete = true;
- propose_pending();
- return;
- }
-
- // osds that went up or down
- set<int> need_check_down_pg_osds;
-
- // apply latest map(s)
- epoch = std::max(epoch, osdmap.get_epoch());
- for (epoch_t e = pg_map.last_osdmap_epoch+1;
- e <= epoch;
- e++) {
- dout(10) << __func__ << " applying osdmap e" << e << " to pg_map" << dendl;
- bufferlist bl;
- int err = mon->osdmon()->get_version(e, bl);
- assert(err == 0);
-
- assert(bl.length());
- OSDMap::Incremental inc(bl);
-
- PGMapUpdater::check_osd_map(inc, &need_check_down_pg_osds,
- &last_osd_report, &pg_map, &pending_inc);
- }
-
- assert(pg_map.last_osdmap_epoch < epoch);
- pending_inc.osdmap_epoch = epoch;
- PGMapUpdater::update_creating_pgs(osdmap, pg_map, &pending_inc);
- PGMapUpdater::register_new_pgs(osdmap, pg_map, &pending_inc);
-
- PGMapUpdater::check_down_pgs(osdmap, pg_map, check_all_pgs,
- need_check_down_pg_osds, &pending_inc);
- check_all_pgs = false;
-
- propose_pending();
-}
-
-epoch_t PGMonitor::send_pg_creates(int osd, Connection *con, epoch_t next)
-{
- dout(30) << __func__ << " " << pg_map.creating_pgs_by_osd_epoch << dendl;
- map<int, map<epoch_t, set<pg_t> > >::iterator p =
- pg_map.creating_pgs_by_osd_epoch.find(osd);
- if (p == pg_map.creating_pgs_by_osd_epoch.end())
- return next;
-
- assert(p->second.size() > 0);
-
- MOSDPGCreate *m = NULL;
- epoch_t last = 0;
- for (map<epoch_t, set<pg_t> >::iterator q = p->second.lower_bound(next);
- q != p->second.end();
- ++q) {
- dout(20) << __func__ << " osd." << osd << " from " << next
- << " : epoch " << q->first << " " << q->second.size() << " pgs"
- << dendl;
- last = q->first;
- for (set<pg_t>::iterator r = q->second.begin(); r != q->second.end(); ++r) {
- pg_stat_t &st = pg_map.pg_stat[*r];
- if (!m)
- m = new MOSDPGCreate(pg_map.last_osdmap_epoch);
- m->mkpg[*r] = pg_create_t(st.created,
- st.parent,
- st.parent_split_bits);
- // Need the create time from the monitor using its clock to set
- // last_scrub_stamp upon pg creation.
- m->ctimes[*r] = pg_map.pg_stat[*r].last_scrub_stamp;
- }
- }
- if (!m) {
- dout(20) << "send_pg_creates osd." << osd << " from " << next
- << " has nothing to send" << dendl;
- return next;
- }
-
- con->send_message(m);
-
- // sub is current through last + 1
- return last + 1;
-}
-
-bool PGMonitor::preprocess_command(MonOpRequestRef op)
-{
- op->mark_pgmon_event(__func__);
- MMonCommand *m = static_cast<MMonCommand*>(op->get_req());
- int r = -1;
- bufferlist rdata;
- stringstream ss, ds;
-
- if (m->fsid != mon->monmap->fsid) {
- dout(0) << __func__ << " drop message on fsid " << m->fsid << " != "
- << mon->monmap->fsid << " for " << *m << dendl;
- return true;
- }
-
- map<string, cmd_vartype> cmdmap;
- if (!cmdmap_from_json(m->cmd, &cmdmap, ss)) {
- // ss has reason for failure
- string rs = ss.str();
- mon->reply_command(op, -EINVAL, rs, rdata, get_last_committed());
- return true;
- }
-
- string prefix;
- cmd_getval(g_ceph_context, cmdmap, "prefix", prefix);
-
- MonSession *session = m->get_session();
- if (!session) {
- mon->reply_command(op, -EACCES, "access denied", rdata, get_last_committed());
- return true;
- }
-
- string format;
- cmd_getval(g_ceph_context, cmdmap, "format", format, string("plain"));
- boost::scoped_ptr<Formatter> f(Formatter::create(format));
-
- if (prefix == "pg scrub" ||
- prefix == "pg repair" ||
- prefix == "pg deep-scrub") {
- string scrubop = prefix.substr(3, string::npos);
- pg_t pgid;
- string pgidstr;
- cmd_getval(g_ceph_context, cmdmap, "pgid", pgidstr);
- if (!pgid.parse(pgidstr.c_str())) {
- ss << "invalid pgid '" << pgidstr << "'";
- r = -EINVAL;
- goto reply;
- }
- if (!pg_map.pg_stat.count(pgid)) {
- ss << "pg " << pgid << " dne";
- r = -ENOENT;
- goto reply;
- }
- int osd = pg_map.pg_stat[pgid].acting_primary;
- if (osd == -1) {
- ss << "pg " << pgid << " has no primary osd";
- r = -EAGAIN;
- goto reply;
- }
- if (!mon->osdmon()->osdmap.is_up(osd)) {
- ss << "pg " << pgid << " primary osd." << osd << " not up";
- r = -EAGAIN;
- goto reply;
- }
- vector<pg_t> pgs(1);
- pgs[0] = pgid;
- mon->try_send_message(new MOSDScrub(mon->monmap->fsid, pgs,
- scrubop == "repair",
- scrubop == "deep-scrub"),
- mon->osdmon()->osdmap.get_inst(osd));
- ss << "instructing pg " << pgid << " on osd." << osd << " to " << scrubop;
- r = 0;
- } else {
- r = process_pg_map_command(prefix, cmdmap, pg_map, mon->osdmon()->osdmap,
- f.get(), &ss, &rdata);
- }
-
- if (r == -EOPNOTSUPP)
- return false;
-
-reply:
- string rs;
- getline(ss, rs);
- rdata.append(ds);
- mon->reply_command(op, r, rs, rdata, get_last_committed());
- return true;
-}
-
-bool PGMonitor::prepare_command(MonOpRequestRef op)
-{
- op->mark_pgmon_event(__func__);
- MMonCommand *m = static_cast<MMonCommand*>(op->get_req());
- if (m->fsid != mon->monmap->fsid) {
- dout(0) << __func__ << " drop message on fsid " << m->fsid << " != "
- << mon->monmap->fsid << " for " << *m << dendl;
- return true;
- }
- stringstream ss;
- pg_t pgid;
- epoch_t epoch = mon->osdmon()->osdmap.get_epoch();
- int r = 0;
- string rs;
-
- map<string, cmd_vartype> cmdmap;
- if (!cmdmap_from_json(m->cmd, &cmdmap, ss)) {
- // ss has reason for failure
- string rs = ss.str();
- mon->reply_command(op, -EINVAL, rs, get_last_committed());
- return true;
- }
-
- string prefix;
- cmd_getval(g_ceph_context, cmdmap, "prefix", prefix);
-
- MonSession *session = m->get_session();
- if (!session) {
- mon->reply_command(op, -EACCES, "access denied", get_last_committed());
- return true;
- }
-
- if (prefix == "pg force_create_pg") {
- string pgidstr;
- cmd_getval(g_ceph_context, cmdmap, "pgid", pgidstr);
- if (!pgid.parse(pgidstr.c_str())) {
- ss << "pg " << pgidstr << " invalid";
- r = -EINVAL;
- goto reply;
- }
- if (!pg_map.pg_stat.count(pgid)) {
- ss << "pg " << pgid << " dne";
- r = -ENOENT;
- goto reply;
- }
- if (pg_map.creating_pgs.count(pgid)) {
- ss << "pg " << pgid << " already creating";
- r = 0;
- goto reply;
- }
- {
- PGMapUpdater::register_pg(
- mon->osdmon()->osdmap,
- pgid,
- epoch,
- true,
- pg_map,
- &pending_inc);
- }
- ss << "pg " << pgidstr << " now creating, ok";
- goto update;
- } else if (prefix == "pg force-recovery" ||
- prefix == "pg force-backfill" ||
- prefix == "pg cancel-force-recovery" ||
- prefix == "pg cancel-force-backfill") {
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- ss << "you must complete the upgrade and 'ceph osd require-osd-release "
- << "luminous' before using forced recovery";
- r = -EPERM;
- goto reply;
- }
- } else if (prefix == "pg set_full_ratio" ||
- prefix == "pg set_nearfull_ratio") {
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- ss << "please use the new luminous interfaces"
- << " ('osd set-full-ratio' and 'osd set-nearfull-ratio')";
- r = -EPERM;
- goto reply;
- }
- double n;
- if (!cmd_getval(g_ceph_context, cmdmap, "ratio", n)) {
- ss << "unable to parse 'ratio' value '"
- << cmd_vartype_stringify(cmdmap["who"]) << "'";
- r = -EINVAL;
- goto reply;
- }
- string op = prefix.substr(3, string::npos);
- if (op == "set_full_ratio")
- pending_inc.full_ratio = n;
- else if (op == "set_nearfull_ratio")
- pending_inc.nearfull_ratio = n;
- goto update;
- } else {
- r = -EINVAL;
- goto reply;
- }
-
-reply:
- getline(ss, rs);
- if (r < 0 && rs.length() == 0)
- rs = cpp_strerror(r);
- mon->reply_command(op, r, rs, get_last_committed());
- return false;
-
-update:
- getline(ss, rs);
- wait_for_finished_proposal(op, new Monitor::C_Command(
- mon, op, r, rs, get_last_committed() + 1));
- return true;
-}
-
-void PGMonitor::get_health(list<pair<health_status_t,string> >& summary,
- list<pair<health_status_t,string> > *detail,
- CephContext *cct) const
-{
- // legacy pre-luminous full/nearfull
- if (mon->osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
- check_full_osd_health(summary, detail, pg_map.full_osds, "full",
- HEALTH_ERR);
- check_full_osd_health(summary, detail, pg_map.nearfull_osds, "near full",
- HEALTH_WARN);
- pg_map.get_health(cct, mon->osdmon()->osdmap, summary, detail);
- }
-}
-
-void PGMonitor::check_full_osd_health(list<pair<health_status_t,string> >& summary,
- list<pair<health_status_t,string> > *detail,
- const mempool::pgmap::set<int>& s, const char *desc,
- health_status_t sev) const
-{
- if (!s.empty()) {
- ostringstream ss;
- ss << s.size() << " " << desc << " osd(s)";
- summary.push_back(make_pair(sev, ss.str()));
- if (detail) {
- for (set<int>::const_iterator p = s.begin(); p != s.end(); ++p) {
- ostringstream ss;
- const osd_stat_t& os = pg_map.osd_stat.find(*p)->second;
- int ratio = (int)(((float)os.kb_used) / (float) os.kb * 100.0);
- ss << "osd." << *p << " is " << desc << " at " << ratio << "%";
- detail->push_back(make_pair(sev, ss.str()));
- }
- }
- }
-}
-
-void PGMonitor::check_subs()
-{
- if (mon->osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
- return;
- }
-
- dout(10) << __func__ << dendl;
- const string type = "osd_pg_creates";
-
- mon->with_session_map([this, &type](const MonSessionMap& session_map) {
- if (mon->session_map.subs.count(type) == 0)
- return;
-
- auto p = mon->session_map.subs[type]->begin();
- while (!p.end()) {
- Subscription *sub = *p;
- ++p;
- dout(20) << __func__ << " .. " << sub->session->inst << dendl;
- check_sub(sub);
- }
- });
-}
-
-bool PGMonitor::check_sub(Subscription *sub)
-{
- OSDMap& osdmap = mon->osdmon()->osdmap;
- if (sub->type == "osd_pg_creates") {
- // only send these if the OSD is up. we will check_subs() when they do
- // come up so they will get the creates then.
- if (sub->session->inst.name.is_osd() &&
- osdmap.is_up(sub->session->inst.name.num())) {
- sub->next = send_pg_creates(sub->session->inst.name.num(),
- sub->session->con.get(),
- sub->next);
- }
- }
- return true;
-}
-
-class PGMonStatService : public MonPGStatService, public PGMapStatService {
- PGMonitor *pgmon;
-public:
- PGMonStatService(const PGMap& o, PGMonitor *pgm)
- : MonPGStatService(), PGMapStatService(o), pgmon(pgm) {}
-
-
- bool is_readable() const override { return pgmon->is_readable(); }
-
- unsigned maybe_add_creating_pgs(epoch_t scan_epoch,
- const mempool::osdmap::map<int64_t,pg_pool_t>& pools,
- creating_pgs_t *pending_creates) const override
- {
- if (pgmap.last_pg_scan < scan_epoch) {
- return 0;
- }
- unsigned added = 0;
- for (auto& pgid : pgmap.creating_pgs) {
- if (!pools.count(pgid.pool())) {
- continue;
- }
- auto st = pgmap.pg_stat.find(pgid);
- assert(st != pgmap.pg_stat.end());
- auto created = make_pair(st->second.created,
- st->second.last_scrub_stamp);
- // no need to add the pg, if it already exists in creating_pgs
- if (pending_creates->pgs.emplace(pgid, created).second) {
- added++;
- }
- }
- return added;
- }
- void maybe_trim_creating_pgs(creating_pgs_t *creates) const override {
- auto p = creates->pgs.begin();
- while (p != creates->pgs.end()) {
- auto q = pgmap.pg_stat.find(p->first);
- if (q != pgmap.pg_stat.end() &&
- !(q->second.state & PG_STATE_CREATING)) {
- p = creates->pgs.erase(p);
- creates->created_pools.insert(q->first.pool());
- } else {
- ++p;
- }
- }
- }
- void dump_info(Formatter *f) const override {
- f->dump_object("pgmap", pgmap);
- f->dump_unsigned("pgmap_first_committed", pgmon->get_first_committed());
- f->dump_unsigned("pgmap_last_committed", pgmon->get_last_committed());
- }
- int process_pg_command(const string& prefix,
- const map<string,cmd_vartype>& cmdmap,
- const OSDMap& osdmap,
- Formatter *f,
- stringstream *ss,
- bufferlist *odata) const override {
- return process_pg_map_command(prefix, cmdmap, pgmap, osdmap, f, ss, odata);
- }
-
- int reweight_by_utilization(const OSDMap &osd_map,
- int oload,
- double max_changef,
- int max_osds,
- bool by_pg, const set<int64_t> *pools,
- bool no_increasing,
- mempool::osdmap::map<int32_t, uint32_t>* new_weights,
- std::stringstream *ss,
- std::string *out_str,
- Formatter *f) const override {
- return reweight::by_utilization(osd_map, pgmap, oload, max_changef,
- max_osds, by_pg, pools, no_increasing,
- new_weights, ss, out_str, f);
- }
-};
-
-MonPGStatService *PGMonitor::get_pg_stat_service()
-{
- if (!pgservice) {
- pgservice.reset(new PGMonStatService(pg_map, this));
- }
- return pgservice.get();
-}
-
-PGMonitor::PGMonitor(Monitor *mn, Paxos *p, const string& service_name)
- : PaxosService(mn, p, service_name),
- pgmap_meta_prefix("pgmap_meta"),
- pgmap_pg_prefix("pgmap_pg"),
- pgmap_osd_prefix("pgmap_osd")
-{}
-
-PGMonitor::~PGMonitor() = default;
Code Review / stor4nfv.git / blobdiff