#include "common/errno.h"
#include "common/ceph_argparse.h"
+#include "common/ceph_time.h"
#include "common/version.h"
#include "common/io_priority.h"
+#include "common/pick_address.h"
#include "os/ObjectStore.h"
#ifdef HAVE_LIBFUSE
#include "messages/MLog.h"
#include "messages/MGenericMessage.h"
-#include "messages/MPing.h"
#include "messages/MOSDPing.h"
#include "messages/MOSDFailure.h"
#include "messages/MOSDMarkMeDown.h"
#include "messages/MOSDPGBackfill.h"
#include "messages/MBackfillReserve.h"
#include "messages/MRecoveryReserve.h"
+#include "messages/MOSDForceRecovery.h"
#include "messages/MOSDECSubOpWrite.h"
#include "messages/MOSDECSubOpWriteReply.h"
#include "messages/MOSDECSubOpRead.h"
#undef dout_prefix
#define dout_prefix _prefix(_dout, whoami, get_osdmap_epoch())
+
const double OSD::OSD_TICK_INTERVAL = 1.0;
static ostream& _prefix(std::ostream* _dout, int whoami, epoch_t epoch) {
return *_dout << "osd." << whoami << " " << epoch << " ";
}
-void PGQueueable::RunVis::operator()(const OpRequestRef &op) {
- return osd->dequeue_op(pg, op, handle);
-}
-
-void PGQueueable::RunVis::operator()(const PGSnapTrim &op) {
- return pg->snap_trimmer(op.epoch_queued);
-}
-
-void PGQueueable::RunVis::operator()(const PGScrub &op) {
- return pg->scrub(op.epoch_queued, handle);
-}
-
-void PGQueueable::RunVis::operator()(const PGRecovery &op) {
- return osd->do_recovery(pg.get(), op.epoch_queued, op.reserved_pushes, handle);
-}
-
//Initial features in new superblock.
//Features here are also automatically upgraded
CompatSet OSD::get_osd_initial_compat_set() {
ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_PGMETA);
ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_MISSING);
ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_FASTINFO);
+ ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_RECOVERY_DELETES);
return CompatSet(ceph_osd_feature_compat, ceph_osd_feature_ro_compat,
ceph_osd_feature_incompat);
}
recovery_sleep_lock("OSDService::recovery_sleep_lock"),
recovery_sleep_timer(cct, recovery_sleep_lock, false),
reserver_finisher(cct),
- local_reserver(&reserver_finisher, cct->_conf->osd_max_backfills,
+ local_reserver(cct, &reserver_finisher, cct->_conf->osd_max_backfills,
cct->_conf->osd_min_recovery_priority),
- remote_reserver(&reserver_finisher, cct->_conf->osd_max_backfills,
+ remote_reserver(cct, &reserver_finisher, cct->_conf->osd_max_backfills,
cct->_conf->osd_min_recovery_priority),
pg_temp_lock("OSDService::pg_temp_lock"),
snap_sleep_lock("OSDService::snap_sleep_lock"),
scrub_sleep_lock("OSDService::scrub_sleep_lock"),
scrub_sleep_timer(
osd->client_messenger->cct, scrub_sleep_lock, false /* relax locking */),
- snap_reserver(&reserver_finisher,
+ snap_reserver(cct, &reserver_finisher,
cct->_conf->osd_max_trimming_pgs),
recovery_lock("OSDService::recovery_lock"),
recovery_ops_active(0),
agent_lock.Unlock();
}
+void OSDService::request_osdmap_update(epoch_t e)
+{
+ osd->osdmap_subscribe(e, false);
+}
+
class AgentTimeoutCB : public Context {
PGRef pg;
public:
return full_ratio;
}
-void OSDService::check_full_status(const osd_stat_t &osd_stat)
+void OSDService::check_full_status(float ratio)
{
Mutex::Locker l(full_status_lock);
- float ratio = ((float)osd_stat.kb_used) / ((float)osd_stat.kb);
cur_ratio = ratio;
// The OSDMap ratios take precendence. So if the failsafe is .95 and
dout(10) << __func__ << " " << get_full_state_name(cur_state)
<< " -> " << get_full_state_name(new_state) << dendl;
if (new_state == FAILSAFE) {
- clog->error() << "failsafe engaged, dropping updates, now "
+ clog->error() << "full status failsafe engaged, dropping updates, now "
<< (int)roundf(ratio * 100) << "% full";
} else if (cur_state == FAILSAFE) {
- clog->error() << "failsafe disengaged, no longer dropping updates, now "
- << (int)roundf(ratio * 100) << "% full";
+ clog->error() << "full status failsafe disengaged, no longer dropping "
+ << "updates, now " << (int)roundf(ratio * 100) << "% full";
}
cur_state = new_state;
}
injectfull = count;
}
-void OSDService::update_osd_stat(vector<int>& hb_peers)
+osd_stat_t OSDService::set_osd_stat(const struct store_statfs_t &stbuf,
+ vector<int>& hb_peers,
+ int num_pgs)
{
- Mutex::Locker lock(stat_lock);
+ uint64_t bytes = stbuf.total;
+ uint64_t used = bytes - stbuf.available;
+ uint64_t avail = stbuf.available;
- osd_stat.hb_peers.swap(hb_peers);
+ osd->logger->set(l_osd_stat_bytes, bytes);
+ osd->logger->set(l_osd_stat_bytes_used, used);
+ osd->logger->set(l_osd_stat_bytes_avail, avail);
- osd->op_tracker.get_age_ms_histogram(&osd_stat.op_queue_age_hist);
+ {
+ Mutex::Locker l(stat_lock);
+ osd_stat.hb_peers.swap(hb_peers);
+ osd->op_tracker.get_age_ms_histogram(&osd_stat.op_queue_age_hist);
+ osd_stat.kb = bytes >> 10;
+ osd_stat.kb_used = used >> 10;
+ osd_stat.kb_avail = avail >> 10;
+ osd_stat.num_pgs = num_pgs;
+ return osd_stat;
+ }
+}
- // fill in osd stats too
+void OSDService::update_osd_stat(vector<int>& hb_peers)
+{
+ // load osd stats first
struct store_statfs_t stbuf;
int r = osd->store->statfs(&stbuf);
if (r < 0) {
return;
}
- uint64_t bytes = stbuf.total;
- uint64_t used = bytes - stbuf.available;
- uint64_t avail = stbuf.available;
-
- osd_stat.kb = bytes >> 10;
- osd_stat.kb_used = used >> 10;
- osd_stat.kb_avail = avail >> 10;
-
- osd->logger->set(l_osd_stat_bytes, bytes);
- osd->logger->set(l_osd_stat_bytes_used, used);
- osd->logger->set(l_osd_stat_bytes_avail, avail);
-
- dout(20) << "update_osd_stat " << osd_stat << dendl;
-
- check_full_status(osd_stat);
+ auto new_stat = set_osd_stat(stbuf, hb_peers, osd->get_num_pgs());
+ dout(20) << "update_osd_stat " << new_stat << dendl;
+ assert(new_stat.kb);
+ float ratio = ((float)new_stat.kb_used) / ((float)new_stat.kb);
+ check_full_status(ratio);
}
bool OSDService::check_osdmap_full(const set<pg_shard_t> &missing_on)
}
-void OSDService::queue_want_pg_temp(pg_t pgid, vector<int>& want)
+void OSDService::queue_want_pg_temp(pg_t pgid,
+ const vector<int>& want,
+ bool forced)
{
Mutex::Locker l(pg_temp_lock);
- map<pg_t,vector<int> >::iterator p = pg_temp_pending.find(pgid);
+ auto p = pg_temp_pending.find(pgid);
if (p == pg_temp_pending.end() ||
- p->second != want) {
- pg_temp_wanted[pgid] = want;
+ p->second.acting != want ||
+ forced) {
+ pg_temp_wanted[pgid] = pg_temp_t{want, forced};
}
}
void OSDService::_sent_pg_temp()
{
- for (map<pg_t,vector<int> >::iterator p = pg_temp_wanted.begin();
- p != pg_temp_wanted.end();
- ++p)
- pg_temp_pending[p->first] = p->second;
+ pg_temp_pending.insert(make_move_iterator(begin(pg_temp_wanted)),
+ make_move_iterator(end(pg_temp_wanted)));
pg_temp_wanted.clear();
}
<< pg_temp_wanted.size() << dendl;
}
+std::ostream& operator<<(std::ostream& out,
+ const OSDService::pg_temp_t& pg_temp)
+{
+ out << pg_temp.acting;
+ if (pg_temp.forced) {
+ out << " (forced)";
+ }
+ return out;
+}
+
void OSDService::send_pg_temp()
{
Mutex::Locker l(pg_temp_lock);
if (pg_temp_wanted.empty())
return;
dout(10) << "send_pg_temp " << pg_temp_wanted << dendl;
- MOSDPGTemp *m = new MOSDPGTemp(osdmap->get_epoch());
- m->pg_temp = pg_temp_wanted;
- monc->send_mon_message(m);
+ MOSDPGTemp *ms[2] = {nullptr, nullptr};
+ for (auto& pg_temp : pg_temp_wanted) {
+ auto& m = ms[pg_temp.second.forced];
+ if (!m) {
+ m = new MOSDPGTemp(osdmap->get_epoch());
+ m->forced = pg_temp.second.forced;
+ }
+ m->pg_temp.emplace(pg_temp.first,
+ pg_temp.second.acting);
+ }
+ for (auto m : ms) {
+ if (m) {
+ monc->send_mon_message(m);
+ }
+ }
_sent_pg_temp();
}
void OSDService::send_pg_created(pg_t pgid)
{
dout(20) << __func__ << dendl;
- monc->send_mon_message(new MOSDPGCreated(pgid));
+ if (osdmap->require_osd_release >= CEPH_RELEASE_LUMINOUS) {
+ monc->send_mon_message(new MOSDPGCreated(pgid));
+ }
}
// --------------------------------------
if (scrubs_pending + scrubs_active < cct->_conf->osd_max_scrubs) {
dout(20) << __func__ << " " << scrubs_pending << " -> " << (scrubs_pending+1)
- << " (max " << cct->_conf->osd_max_scrubs << ", active " << scrubs_active << ")" << dendl;
+ << " (max " << cct->_conf->osd_max_scrubs << ", active " << scrubs_active
+ << ")" << dendl;
can_inc = true;
} else {
- dout(20) << __func__ << scrubs_pending << " + " << scrubs_active << " active >= max " << cct->_conf->osd_max_scrubs << dendl;
+ dout(20) << __func__ << " " << scrubs_pending << " + " << scrubs_active
+ << " active >= max " << cct->_conf->osd_max_scrubs << dendl;
}
return can_inc;
MOSDMap *OSDService::build_incremental_map_msg(epoch_t since, epoch_t to,
OSDSuperblock& sblock)
{
- MOSDMap *m = new MOSDMap(monc->get_fsid());
+ MOSDMap *m = new MOSDMap(monc->get_fsid(),
+ osdmap->get_encoding_features());
m->oldest_map = max_oldest_map;
m->newest_map = sblock.newest_map;
OSDSuperblock sblock(get_superblock());
if (since < sblock.oldest_map) {
// just send latest full map
- MOSDMap *m = new MOSDMap(monc->get_fsid());
+ MOSDMap *m = new MOSDMap(monc->get_fsid(),
+ osdmap->get_encoding_features());
m->oldest_map = max_oldest_map;
m->newest_map = sblock.newest_map;
get_map_bl(to, m->maps[to]);
ret = store->mkfs();
if (ret) {
- derr << "OSD::mkfs: ObjectStore::mkfs failed with error " << ret << dendl;
+ derr << "OSD::mkfs: ObjectStore::mkfs failed with error "
+ << cpp_strerror(ret) << dendl;
goto free_store;
}
ret = store->mount();
if (ret) {
- derr << "OSD::mkfs: couldn't mount ObjectStore: error " << ret << dendl;
+ derr << "OSD::mkfs: couldn't mount ObjectStore: error "
+ << cpp_strerror(ret) << dendl;
goto free_store;
}
ret = store->apply_transaction(osr.get(), std::move(t));
if (ret) {
derr << "OSD::mkfs: error while writing OSD_SUPERBLOCK_GOBJECT: "
- << "apply_transaction returned " << ret << dendl;
+ << "apply_transaction returned " << cpp_strerror(ret) << dendl;
goto umount_store;
}
}
waiter.wait();
}
- ret = write_meta(store, sb.cluster_fsid, sb.osd_fsid, whoami);
+ ret = write_meta(cct, store, sb.cluster_fsid, sb.osd_fsid, whoami);
if (ret) {
- derr << "OSD::mkfs: failed to write fsid file: error " << ret << dendl;
+ derr << "OSD::mkfs: failed to write fsid file: error "
+ << cpp_strerror(ret) << dendl;
goto umount_store;
}
return ret;
}
-int OSD::write_meta(ObjectStore *store, uuid_d& cluster_fsid, uuid_d& osd_fsid, int whoami)
+int OSD::write_meta(CephContext *cct, ObjectStore *store, uuid_d& cluster_fsid, uuid_d& osd_fsid, int whoami)
{
char val[80];
int r;
if (r < 0)
return r;
+ string key = cct->_conf->get_val<string>("key");
+ if (key.size()) {
+ r = store->write_meta("osd_key", key);
+ if (r < 0)
+ return r;
+ } else {
+ string keyfile = cct->_conf->get_val<string>("keyfile");
+ if (!keyfile.empty()) {
+ bufferlist keybl;
+ string err;
+ if (keyfile == "-") {
+ static_assert(1024 * 1024 >
+ (sizeof(CryptoKey) - sizeof(bufferptr) +
+ sizeof(__u16) + 16 /* AES_KEY_LEN */ + 3 - 1) / 3. * 4.,
+ "1MB should be enough for a base64 encoded CryptoKey");
+ r = keybl.read_fd(STDIN_FILENO, 1024 * 1024);
+ } else {
+ r = keybl.read_file(keyfile.c_str(), &err);
+ }
+ if (r < 0) {
+ derr << __func__ << " failed to read keyfile " << keyfile << ": "
+ << err << ": " << cpp_strerror(r) << dendl;
+ return r;
+ }
+ r = store->write_meta("osd_key", keybl.to_str());
+ if (r < 0)
+ return r;
+ }
+ }
+
r = store->write_meta("ready", "ready");
if (r < 0)
return r;
disk_tp(cct, "OSD::disk_tp", "tp_osd_disk", cct->_conf->osd_disk_threads, "osd_disk_threads"),
command_tp(cct, "OSD::command_tp", "tp_osd_cmd", 1),
session_waiting_lock("OSD::session_waiting_lock"),
+ osdmap_subscribe_lock("OSD::osdmap_subscribe_lock"),
heartbeat_lock("OSD::heartbeat_lock"),
heartbeat_stop(false),
heartbeat_need_update(true),
} else if (admin_command == "flush_journal") {
store->flush_journal();
} else if (admin_command == "dump_ops_in_flight" ||
- admin_command == "ops") {
- if (!op_tracker.dump_ops_in_flight(f)) {
- ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
- Please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
- }
- } else if (admin_command == "dump_blocked_ops") {
- if (!op_tracker.dump_ops_in_flight(f, true)) {
- ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
- Please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
- }
- } else if (admin_command == "dump_historic_ops") {
- if (!op_tracker.dump_historic_ops(f, false)) {
- ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
- Please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
- }
- } else if (admin_command == "dump_historic_ops_by_duration") {
- if (!op_tracker.dump_historic_ops(f, true)) {
- ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
- Please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
- }
- } else if (admin_command == "dump_historic_slow_ops") {
- if (!op_tracker.dump_historic_slow_ops(f)) {
- ss << "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
- Please enable \"osd_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
+ admin_command == "ops" ||
+ admin_command == "dump_blocked_ops" ||
+ admin_command == "dump_historic_ops" ||
+ admin_command == "dump_historic_ops_by_duration" ||
+ admin_command == "dump_historic_slow_ops") {
+
+ const string error_str = "op_tracker tracking is not enabled now, so no ops are tracked currently, \
+even those get stuck. Please enable \"osd_enable_op_tracker\", and the tracker \
+will start to track new ops received afterwards.";
+
+ set<string> filters;
+ vector<string> filter_str;
+ if (cmd_getval(cct, cmdmap, "filterstr", filter_str)) {
+ copy(filter_str.begin(), filter_str.end(),
+ inserter(filters, filters.end()));
+ }
+
+ if (admin_command == "dump_ops_in_flight" ||
+ admin_command == "ops") {
+ if (!op_tracker.dump_ops_in_flight(f, false, filters)) {
+ ss << error_str;
+ }
+ }
+ if (admin_command == "dump_blocked_ops") {
+ if (!op_tracker.dump_ops_in_flight(f, true, filters)) {
+ ss << error_str;
+ }
+ }
+ if (admin_command == "dump_historic_ops") {
+ if (!op_tracker.dump_historic_ops(f, false, filters)) {
+ ss << error_str;
+ }
+ }
+ if (admin_command == "dump_historic_ops_by_duration") {
+ if (!op_tracker.dump_historic_ops(f, true, filters)) {
+ ss << error_str;
+ }
+ }
+ if (admin_command == "dump_historic_slow_ops") {
+ if (!op_tracker.dump_historic_slow_ops(f, filters)) {
+ ss << error_str;
+ }
}
} else if (admin_command == "dump_op_pq_state") {
f->open_object_section("pq");
curmap->get_blacklist(&bl);
for (list<pair<entity_addr_t,utime_t> >::iterator it = bl.begin();
it != bl.end(); ++it) {
- f->open_array_section("entry");
+ f->open_object_section("entry");
f->open_object_section("entity_addr_t");
it->first.dump(f);
f->close_section(); //entity_addr_t
for (list<obj_watch_item_t>::iterator it = watchers.begin();
it != watchers.end(); ++it) {
- f->open_array_section("watch");
+ f->open_object_section("watch");
f->dump_string("namespace", it->obj.nspace);
f->dump_string("object", it->obj.oid.name);
it->wi.name.dump(f);
f->close_section(); //entity_name_t
- f->dump_int("cookie", it->wi.cookie);
- f->dump_int("timeout", it->wi.timeout_seconds);
+ f->dump_unsigned("cookie", it->wi.cookie);
+ f->dump_unsigned("timeout", it->wi.timeout_seconds);
f->open_object_section("entity_addr_t");
it->wi.addr.dump(f);
pg->unlock();
}
f->close_section();
+ } else if (admin_command == "compact") {
+ dout(1) << "triggering manual compaction" << dendl;
+ auto start = ceph::coarse_mono_clock::now();
+ store->compact();
+ auto end = ceph::coarse_mono_clock::now();
+ auto time_span = chrono::duration_cast<chrono::duration<double>>(end - start);
+ dout(1) << "finished manual compaction in "
+ << time_span.count()
+ << " seconds" << dendl;
+ f->open_object_section("compact_result");
+ f->dump_float("elapsed_time", time_span.count());
+ f->close_section();
} else {
assert(0 == "broken asok registration");
}
delete fuse_store;
fuse_store = NULL;
r = ::rmdir(mntpath.c_str());
- if (r < 0)
- r = -errno;
if (r < 0) {
- derr << __func__ << " failed to rmdir " << mntpath << dendl;
+ r = -errno;
+ derr << __func__ << " failed to rmdir " << mntpath << ": "
+ << cpp_strerror(r) << dendl;
return r;
}
return 0;
return get_num_op_shards() * cct->_conf->osd_op_num_threads_per_shard_ssd;
}
+float OSD::get_osd_recovery_sleep()
+{
+ if (cct->_conf->osd_recovery_sleep)
+ return cct->_conf->osd_recovery_sleep;
+ if (!store_is_rotational && !journal_is_rotational)
+ return cct->_conf->osd_recovery_sleep_ssd;
+ else if (store_is_rotational && !journal_is_rotational)
+ return cct->_conf->get_val<double>("osd_recovery_sleep_hybrid");
+ else
+ return cct->_conf->osd_recovery_sleep_hdd;
+}
+
int OSD::init()
{
CompatSet initial, diff;
dout(2) << "init " << dev_path
<< " (looks like " << (store_is_rotational ? "hdd" : "ssd") << ")"
<< dendl;
+ dout(2) << "journal " << journal_path << dendl;
assert(store); // call pre_init() first!
store->set_cache_shards(get_num_op_shards());
derr << "OSD:init: unable to mount object store" << dendl;
return r;
}
+ journal_is_rotational = store->is_journal_rotational();
+ dout(2) << "journal looks like " << (journal_is_rotational ? "hdd" : "ssd")
+ << dendl;
enable_disable_fuse(false);
clear_temp_objects();
+ // initialize osdmap references in sharded wq
+ op_shardedwq.prune_pg_waiters(osdmap, whoami);
+
// load up pgs (as they previously existed)
load_pgs();
update_log_config();
peering_tp.start();
+
+ service.init();
+ service.publish_map(osdmap);
+ service.publish_superblock(superblock);
+ service.max_oldest_map = superblock.oldest_map;
+
osd_op_tp.start();
disk_tp.start();
command_tp.start();
tick_timer_without_osd_lock.add_event_after(cct->_conf->osd_heartbeat_interval, new C_Tick_WithoutOSDLock(this));
}
- service.init();
- service.publish_map(osdmap);
- service.publish_superblock(superblock);
- service.max_oldest_map = superblock.oldest_map;
-
osd_lock.Unlock();
r = monc->authenticate();
if (r < 0) {
+ derr << __func__ << " authentication failed: " << cpp_strerror(r)
+ << dendl;
osd_lock.Lock(); // locker is going to unlock this on function exit
if (is_stopping())
- r = 0;
+ r = 0;
goto monout;
}
derr << "unable to obtain rotating service keys; retrying" << dendl;
++rotating_auth_attempts;
if (rotating_auth_attempts > g_conf->max_rotating_auth_attempts) {
+ derr << __func__ << " wait_auth_rotating timed out" << dendl;
osd_lock.Lock(); // make locker happy
if (!is_stopping()) {
- r = - ETIMEDOUT;
+ r = -ETIMEDOUT;
}
goto monout;
}
r = update_crush_device_class();
if (r < 0) {
+ derr << __func__ << " unable to update_crush_device_class: "
+ << cpp_strerror(r) << dendl;
osd_lock.Lock();
goto monout;
}
r = update_crush_location();
if (r < 0) {
+ derr << __func__ << " unable to update_crush_location: "
+ << cpp_strerror(r) << dendl;
osd_lock.Lock();
goto monout;
}
return 0;
monout:
- mgrc.shutdown();
- monc->shutdown();
+ exit(1);
out:
enable_disable_fuse(true);
"flush the journal to permanent store");
assert(r == 0);
r = admin_socket->register_command("dump_ops_in_flight",
- "dump_ops_in_flight", asok_hook,
+ "dump_ops_in_flight " \
+ "name=filterstr,type=CephString,n=N,req=false",
+ asok_hook,
"show the ops currently in flight");
assert(r == 0);
r = admin_socket->register_command("ops",
- "ops", asok_hook,
+ "ops " \
+ "name=filterstr,type=CephString,n=N,req=false",
+ asok_hook,
"show the ops currently in flight");
assert(r == 0);
r = admin_socket->register_command("dump_blocked_ops",
- "dump_blocked_ops", asok_hook,
+ "dump_blocked_ops " \
+ "name=filterstr,type=CephString,n=N,req=false",
+ asok_hook,
"show the blocked ops currently in flight");
assert(r == 0);
- r = admin_socket->register_command("dump_historic_ops", "dump_historic_ops",
+ r = admin_socket->register_command("dump_historic_ops",
+ "dump_historic_ops " \
+ "name=filterstr,type=CephString,n=N,req=false",
asok_hook,
"show recent ops");
assert(r == 0);
- r = admin_socket->register_command("dump_historic_slow_ops", "dump_historic_slow_ops",
+ r = admin_socket->register_command("dump_historic_slow_ops",
+ "dump_historic_slow_ops " \
+ "name=filterstr,type=CephString,n=N,req=false",
asok_hook,
"show slowest recent ops");
assert(r == 0);
- r = admin_socket->register_command("dump_historic_ops_by_duration", "dump_historic_ops_by_duration",
+ r = admin_socket->register_command("dump_historic_ops_by_duration",
+ "dump_historic_ops_by_duration " \
+ "name=filterstr,type=CephString,n=N,req=false",
asok_hook,
"show slowest recent ops, sorted by duration");
assert(r == 0);
"show recent state history");
assert(r == 0);
+ r = admin_socket->register_command("compact", "compact",
+ asok_hook,
+ "Commpact object store's omap."
+ " WARNING: Compaction probably slows your requests");
+ assert(r == 0);
+
test_ops_hook = new TestOpsSocketHook(&(this->service), this->store);
// Note: pools are CephString instead of CephPoolname because
// these commands traditionally support both pool names and numbers
};
+ // All the basic OSD operation stats are to be considered useful
+ osd_plb.set_prio_default(PerfCountersBuilder::PRIO_USEFUL);
+
osd_plb.add_u64(
l_osd_op_wip, "op_wip",
"Replication operations currently being processed (primary)");
l_osd_op_rw_prepare_lat, "op_rw_prepare_latency",
"Latency of read-modify-write operations (excluding queue time and wait for finished)");
+ // Now we move on to some more obscure stats, revert to assuming things
+ // are low priority unless otherwise specified.
+ osd_plb.set_prio_default(PerfCountersBuilder::PRIO_DEBUGONLY);
+
+ osd_plb.add_time_avg(l_osd_op_before_queue_op_lat, "op_before_queue_op_lat",
+ "Latency of IO before calling queue(before really queue into ShardedOpWq)"); // client io before queue op_wq latency
+ osd_plb.add_time_avg(l_osd_op_before_dequeue_op_lat, "op_before_dequeue_op_lat",
+ "Latency of IO before calling dequeue_op(already dequeued and get PG lock)"); // client io before dequeue_op latency
+
osd_plb.add_u64_counter(
l_osd_sop, "subop", "Suboperations");
osd_plb.add_u64_counter(
osd_plb.add_u64(
l_osd_pg_stray, "numpg_stray",
"Placement groups ready to be deleted from this osd");
+ osd_plb.add_u64(
+ l_osd_pg_removing, "numpg_removing",
+ "Placement groups queued for local deletion", "pgsr",
+ PerfCountersBuilder::PRIO_USEFUL);
osd_plb.add_u64(
l_osd_hb_to, "heartbeat_to_peers", "Heartbeat (ping) peers we send to");
osd_plb.add_u64_counter(l_osd_map, "map_messages", "OSD map messages");
l_osd_map_bl_cache_miss, "osd_map_bl_cache_miss",
"OSDMap buffer cache misses");
- osd_plb.add_u64(l_osd_stat_bytes, "stat_bytes", "OSD size");
- osd_plb.add_u64(l_osd_stat_bytes_used, "stat_bytes_used", "Used space");
+ osd_plb.add_u64(
+ l_osd_stat_bytes, "stat_bytes", "OSD size", "size",
+ PerfCountersBuilder::PRIO_USEFUL);
+ osd_plb.add_u64(
+ l_osd_stat_bytes_used, "stat_bytes_used", "Used space", "used",
+ PerfCountersBuilder::PRIO_USEFUL);
osd_plb.add_u64(l_osd_stat_bytes_avail, "stat_bytes_avail", "Available space");
osd_plb.add_u64_counter(
set_state(STATE_STOPPING);
// Debugging
- cct->_conf->set_val("debug_osd", "100");
- cct->_conf->set_val("debug_journal", "100");
- cct->_conf->set_val("debug_filestore", "100");
- cct->_conf->set_val("debug_ms", "100");
- cct->_conf->apply_changes(NULL);
+ if (cct->_conf->get_val<bool>("osd_debug_shutdown")) {
+ cct->_conf->set_val("debug_osd", "100");
+ cct->_conf->set_val("debug_journal", "100");
+ cct->_conf->set_val("debug_filestore", "100");
+ cct->_conf->set_val("debug_bluestore", "100");
+ cct->_conf->set_val("debug_ms", "100");
+ cct->_conf->apply_changes(NULL);
+ }
// stop MgrClient earlier as it's more like an internal consumer of OSD
mgrc.shutdown();
cct->get_admin_socket()->unregister_command("dump_watchers");
cct->get_admin_socket()->unregister_command("dump_reservations");
cct->get_admin_socket()->unregister_command("get_latest_osdmap");
+ cct->get_admin_socket()->unregister_command("heap");
cct->get_admin_socket()->unregister_command("set_heap_property");
cct->get_admin_socket()->unregister_command("get_heap_property");
cct->get_admin_socket()->unregister_command("dump_objectstore_kv_stats");
+ cct->get_admin_socket()->unregister_command("dump_scrubs");
cct->get_admin_socket()->unregister_command("calc_objectstore_db_histogram");
cct->get_admin_socket()->unregister_command("flush_store_cache");
cct->get_admin_socket()->unregister_command("dump_pgstate_history");
+ cct->get_admin_socket()->unregister_command("compact");
delete asok_hook;
asok_hook = NULL;
cct->get_admin_socket()->unregister_command("injectdataerr");
cct->get_admin_socket()->unregister_command("injectmdataerr");
cct->get_admin_socket()->unregister_command("set_recovery_delay");
+ cct->get_admin_socket()->unregister_command("trigger_scrub");
+ cct->get_admin_socket()->unregister_command("injectfull");
delete test_ops_hook;
test_ops_hook = NULL;
int OSD::update_crush_device_class()
{
+ if (!cct->_conf->osd_class_update_on_start) {
+ dout(10) << __func__ << " osd_class_update_on_start = false" << dendl;
+ return 0;
+ }
+
string device_class;
int r = store->read_meta("crush_device_class", &device_class);
- if (r < 0)
+ if (r < 0 || device_class.empty()) {
+ device_class = store->get_default_device_class();
+ }
+
+ if (device_class.empty()) {
+ dout(20) << __func__ << " no device class stored locally" << dendl;
return 0;
+ }
string cmd =
string("{\"prefix\": \"osd crush set-device-class\", ") +
- string("\"id\": ") + stringify(whoami) + string(", ") +
- string("\"class\": \"") + device_class + string("\"}");
-
- return mon_cmd_maybe_osd_create(cmd);
+ string("\"class\": \"") + device_class + string("\", ") +
+ string("\"ids\": [\"") + stringify(whoami) + string("\"]}");
+
+ r = mon_cmd_maybe_osd_create(cmd);
+ // the above cmd can fail for various reasons, e.g.:
+ // (1) we are connecting to a pre-luminous monitor
+ // (2) user manually specify a class other than
+ // 'ceph-disk prepare --crush-device-class'
+ // simply skip result-checking for now
+ return 0;
}
void OSD::write_superblock(ObjectStore::Transaction& t)
pg->upgrade(store);
}
+ if (pg->dne()) {
+ dout(10) << "load_pgs " << *it << " deleting dne" << dendl;
+ pg->ch = nullptr;
+ service.pg_remove_epoch(pg->pg_id);
+ pg->unlock();
+ {
+ // Delete pg
+ RWLock::WLocker l(pg_map_lock);
+ auto p = pg_map.find(pg->get_pgid());
+ assert(p != pg_map.end() && p->second == pg);
+ dout(20) << __func__ << " removed pg " << pg << " from pg_map" << dendl;
+ pg_map.erase(p);
+ pg->put("PGMap");
+ }
+ recursive_remove_collection(cct, store, pgid, *it);
+ continue;
+ }
+
service.init_splits_between(pg->info.pgid, pg->get_osdmap(), osdmap);
// generate state for PG's current mapping
++i) {
PG *pg = i->second;
+ // Ignore PGs only partially created (DNE)
+ if (pg->info.dne()) {
+ continue;
+ }
+
auto rpib = pg->get_required_past_interval_bounds(
pg->info,
superblock.oldest_map);
ceph_abort();
}
+ const bool is_mon_create =
+ evt->get_event().dynamic_type() == PG::NullEvt::static_type();
+ if (maybe_wait_for_max_pg(pgid, is_mon_create)) {
+ return -EAGAIN;
+ }
// do we need to resurrect a deleting pg?
spg_t resurrected;
PGRef old_pg_state;
}
}
+bool OSD::maybe_wait_for_max_pg(spg_t pgid, bool is_mon_create)
+{
+ const auto max_pgs_per_osd =
+ (cct->_conf->get_val<uint64_t>("mon_max_pg_per_osd") *
+ cct->_conf->get_val<double>("osd_max_pg_per_osd_hard_ratio"));
+
+ RWLock::RLocker pg_map_locker{pg_map_lock};
+ if (pg_map.size() < max_pgs_per_osd) {
+ return false;
+ }
+ lock_guard<mutex> pending_creates_locker{pending_creates_lock};
+ if (is_mon_create) {
+ pending_creates_from_mon++;
+ } else {
+ bool is_primary = osdmap->get_pg_acting_rank(pgid.pgid, whoami) == 0;
+ pending_creates_from_osd.emplace(pgid.pgid, is_primary);
+ }
+ dout(5) << __func__ << " withhold creation of pg " << pgid
+ << ": " << pg_map.size() << " >= "<< max_pgs_per_osd << dendl;
+ return true;
+}
+
+// to re-trigger a peering, we have to twiddle the pg mapping a little bit,
+// see PG::should_restart_peering(). OSDMap::pg_to_up_acting_osds() will turn
+// to up set if pg_temp is empty. so an empty pg_temp won't work.
+static vector<int32_t> twiddle(const vector<int>& acting) {
+ if (acting.size() > 1) {
+ return {acting[0]};
+ } else {
+ vector<int32_t> twiddled(acting.begin(), acting.end());
+ twiddled.push_back(-1);
+ return twiddled;
+ }
+}
+
+void OSD::resume_creating_pg()
+{
+ bool do_sub_pg_creates = false;
+ bool have_pending_creates = false;
+ {
+ const auto max_pgs_per_osd =
+ (cct->_conf->get_val<uint64_t>("mon_max_pg_per_osd") *
+ cct->_conf->get_val<double>("osd_max_pg_per_osd_hard_ratio"));
+ RWLock::RLocker l(pg_map_lock);
+ if (max_pgs_per_osd <= pg_map.size()) {
+ // this could happen if admin decreases this setting before a PG is removed
+ return;
+ }
+ unsigned spare_pgs = max_pgs_per_osd - pg_map.size();
+ lock_guard<mutex> pending_creates_locker{pending_creates_lock};
+ if (pending_creates_from_mon > 0) {
+ do_sub_pg_creates = true;
+ if (pending_creates_from_mon >= spare_pgs) {
+ spare_pgs = pending_creates_from_mon = 0;
+ } else {
+ spare_pgs -= pending_creates_from_mon;
+ pending_creates_from_mon = 0;
+ }
+ }
+ auto pg = pending_creates_from_osd.cbegin();
+ while (spare_pgs > 0 && pg != pending_creates_from_osd.cend()) {
+ dout(20) << __func__ << " pg " << pg->first << dendl;
+ vector<int> acting;
+ osdmap->pg_to_up_acting_osds(pg->first, nullptr, nullptr, &acting, nullptr);
+ service.queue_want_pg_temp(pg->first, twiddle(acting), true);
+ pg = pending_creates_from_osd.erase(pg);
+ do_sub_pg_creates = true;
+ spare_pgs--;
+ }
+ have_pending_creates = (pending_creates_from_mon > 0 ||
+ !pending_creates_from_osd.empty());
+ }
+
+ bool do_renew_subs = false;
+ if (do_sub_pg_creates) {
+ if (monc->sub_want("osd_pg_creates", last_pg_create_epoch, 0)) {
+ dout(4) << __func__ << ": resolicit pg creates from mon since "
+ << last_pg_create_epoch << dendl;
+ do_renew_subs = true;
+ }
+ }
+ version_t start = osdmap->get_epoch() + 1;
+ if (have_pending_creates) {
+ // don't miss any new osdmap deleting PGs
+ if (monc->sub_want("osdmap", start, 0)) {
+ dout(4) << __func__ << ": resolicit osdmap from mon since "
+ << start << dendl;
+ do_renew_subs = true;
+ }
+ } else if (do_sub_pg_creates) {
+ // no need to subscribe the osdmap continuously anymore
+ // once the pgtemp and/or mon_subscribe(pg_creates) is sent
+ if (monc->sub_want_increment("osdmap", start, CEPH_SUBSCRIBE_ONETIME)) {
+ dout(4) << __func__ << ": re-subscribe osdmap(onetime) since"
+ << start << dendl;
+ do_renew_subs = true;
+ }
+ }
+
+ if (do_renew_subs) {
+ monc->renew_subs();
+ }
+
+ service.send_pg_temp();
+}
void OSD::build_initial_pg_history(
spg_t pgid,
&debug);
if (new_interval) {
h->same_interval_since = e;
- }
- if (up != new_up) {
- h->same_up_since = e;
- }
- if (acting_primary != new_acting_primary) {
- h->same_primary_since = e;
+ if (up != new_up) {
+ h->same_up_since = e;
+ }
+ if (acting_primary != new_acting_primary) {
+ h->same_primary_since = e;
+ }
+ if (pgid.pgid.is_split(lastmap->get_pg_num(pgid.pgid.pool()),
+ osdmap->get_pg_num(pgid.pgid.pool()),
+ nullptr)) {
+ h->last_epoch_split = e;
+ }
+ up = new_up;
+ acting = new_acting;
+ up_primary = new_up_primary;
+ acting_primary = new_acting_primary;
}
lastmap = osdmap;
}
}
OSDMapRef curmap = service.get_osdmap();
- assert(curmap);
+ if (!curmap) {
+ heartbeat_lock.Unlock();
+ m->put();
+ return;
+ }
switch (m->op) {
if (is_waiting_for_healthy()) {
start_boot();
+ } else if (is_preboot() &&
+ waiting_for_luminous_mons &&
+ monc->monmap.get_required_features().contains_all(
+ ceph::features::mon::FEATURE_LUMINOUS)) {
+ // mon upgrade finished!
+ start_boot();
}
do_waiters();
tick_timer.add_event_after(OSD_TICK_INTERVAL, new C_Tick(this));
-
- if (is_active()) {
- const auto now = ceph::coarse_mono_clock::now();
- const auto elapsed = now - last_sent_beacon;
- if (chrono::duration_cast<chrono::seconds>(elapsed).count() >
- cct->_conf->osd_beacon_report_interval) {
- send_beacon(now);
- }
- }
}
void OSD::tick_without_osd_lock()
logger->set(l_osd_cached_crc, buffer::get_cached_crc());
logger->set(l_osd_cached_crc_adjusted, buffer::get_cached_crc_adjusted());
logger->set(l_osd_missed_crc, buffer::get_missed_crc());
+ logger->set(l_osd_pg_removing, remove_wq.get_remove_queue_len());
// osd_lock is not being held, which means the OSD state
// might change when doing the monitor report
sched_scrub();
}
service.promote_throttle_recalibrate();
+ resume_creating_pg();
+ bool need_send_beacon = false;
+ const auto now = ceph::coarse_mono_clock::now();
+ {
+ // borrow lec lock to pretect last_sent_beacon from changing
+ Mutex::Locker l{min_last_epoch_clean_lock};
+ const auto elapsed = now - last_sent_beacon;
+ if (chrono::duration_cast<chrono::seconds>(elapsed).count() >
+ cct->_conf->osd_beacon_report_interval) {
+ need_send_beacon = true;
+ }
+ }
+ if (need_send_beacon) {
+ send_beacon(now);
+ }
}
- check_ops_in_flight();
+ mgrc.update_osd_health(get_health_metrics());
service.kick_recovery_queue();
tick_timer_without_osd_lock.add_event_after(OSD_TICK_INTERVAL, new C_Tick_WithoutOSDLock(this));
}
if (pool < 0 && isdigit(poolstr[0]))
pool = atoll(poolstr.c_str());
if (pool < 0) {
- ss << "Invalid pool" << poolstr;
+ ss << "Invalid pool '" << poolstr << "''";
return;
}
}
dout(1) << __func__ << dendl;
set_state(STATE_PREBOOT);
+ waiting_for_luminous_mons = false;
dout(10) << "start_boot - have maps " << superblock.oldest_map
<< ".." << superblock.newest_map << dendl;
C_OSD_GetVersion *c = new C_OSD_GetVersion(this);
// if our map within recent history, try to add ourselves to the osdmap.
if (osdmap->get_epoch() == 0) {
derr << "waiting for initial osdmap" << dendl;
+ } else if (osdmap->is_destroyed(whoami)) {
+ derr << "osdmap says I am destroyed" << dendl;
+ // provide a small margin so we don't livelock seeing if we
+ // un-destroyed ourselves.
+ if (osdmap->get_epoch() > newest - 1) {
+ exit(0);
+ }
} else if (osdmap->test_flag(CEPH_OSDMAP_NOUP) || osdmap->is_noup(whoami)) {
derr << "osdmap NOUP flag is set, waiting for it to clear" << dendl;
} else if (!osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE)) {
ceph::features::mon::FEATURE_LUMINOUS)) {
derr << "monmap REQUIRE_LUMINOUS is NOT set; must upgrade all monitors to "
<< "Luminous or later before Luminous OSDs will boot" << dendl;
+ waiting_for_luminous_mons = true;
} else if (service.need_fullness_update()) {
derr << "osdmap fullness state needs update" << dendl;
send_full_update();
dout(1) << "start_waiting_for_healthy" << dendl;
set_state(STATE_WAITING_FOR_HEALTHY);
last_heartbeat_resample = utime_t();
+
+ // subscribe to osdmap updates, in case our peers really are known to be dead
+ osdmap_subscribe(osdmap->get_epoch() + 1, false);
}
bool OSD::_is_healthy()
{
// config info
(*pm)["osd_data"] = dev_path;
- (*pm)["osd_journal"] = journal_path;
+ if (store->get_type() == "filestore") {
+ // not applicable for bluestore
+ (*pm)["osd_journal"] = journal_path;
+ }
(*pm)["front_addr"] = stringify(client_messenger->get_myaddr());
(*pm)["back_addr"] = stringify(cluster_messenger->get_myaddr());
(*pm)["hb_front_addr"] = stringify(hb_front_server_messenger->get_myaddr());
// backend
(*pm)["osd_objectstore"] = store->get_type();
(*pm)["rotational"] = store_is_rotational ? "1" : "0";
+ (*pm)["journal_rotational"] = journal_is_rotational ? "1" : "0";
+ (*pm)["default_device_class"] = store->get_default_device_class();
store->collect_metadata(pm);
collect_sys_info(pm, cct);
+ std::string front_iface, back_iface;
+ /*
+ pick_iface(cct,
+ CEPH_PICK_ADDRESS_PUBLIC | CEPH_PICK_ADDRESS_CLUSTER,
+ &front_iface, &back_iface);
+ */
+ (*pm)["front_iface"] = pick_iface(cct,
+ client_messenger->get_myaddr().get_sockaddr_storage());
+ (*pm)["back_iface"] = pick_iface(cct,
+ cluster_messenger->get_myaddr().get_sockaddr_storage());
+
dout(10) << __func__ << " " << *pm << dendl;
}
monmap.get_required_features().contains_all(
ceph::features::mon::FEATURE_LUMINOUS)) {
dout(20) << __func__ << " sending" << dendl;
- last_sent_beacon = now;
MOSDBeacon* beacon = nullptr;
{
Mutex::Locker l{min_last_epoch_clean_lock};
beacon = new MOSDBeacon(osdmap->get_epoch(), min_last_epoch_clean);
std::swap(beacon->pgs, min_last_epoch_clean_pgs);
+ last_sent_beacon = now;
}
monc->send_mon_message(beacon);
} else {
"name=injected_args,type=CephString,n=N",
"inject configuration arguments into running OSD",
"osd", "rw", "cli,rest")
+COMMAND("config set " \
+ "name=key,type=CephString name=value,type=CephString",
+ "Set a configuration option at runtime (not persistent)",
+ "osd", "rw", "cli,rest")
COMMAND("cluster_log " \
"name=level,type=CephChoices,strings=error,warning,info,debug " \
"name=message,type=CephString,n=N",
"osd", "r", "cli,rest")
COMMAND("reset_pg_recovery_stats", "reset pg recovery statistics",
"osd", "rw", "cli,rest")
+COMMAND("compact",
+ "compact object store's omap. "
+ "WARNING: Compaction probably slows your requests",
+ "osd", "rw", "cli,rest")
};
void OSD::do_command(Connection *con, ceph_tid_t tid, vector<string>& cmd, bufferlist& data)
r = cct->_conf->injectargs(args, &ss);
osd_lock.Lock();
}
+ else if (prefix == "config set") {
+ std::string key;
+ std::string val;
+ cmd_getval(cct, cmdmap, "key", key);
+ cmd_getval(cct, cmdmap, "value", val);
+ osd_lock.Unlock();
+ r = cct->_conf->set_val(key, val, true, &ss);
+ if (r == 0) {
+ cct->_conf->apply_changes(nullptr);
+ }
+ osd_lock.Lock();
+ }
else if (prefix == "cluster_log") {
vector<string> msg;
cmd_getval(cct, cmdmap, "message", msg);
}
}
+ else if (prefix == "compact") {
+ dout(1) << "triggering manual compaction" << dendl;
+ auto start = ceph::coarse_mono_clock::now();
+ store->compact();
+ auto end = ceph::coarse_mono_clock::now();
+ auto time_span = chrono::duration_cast<chrono::duration<double>>(end - start);
+ dout(1) << "finished manual compaction in "
+ << time_span.count()
+ << " seconds" << dendl;
+ ss << "compacted omap in " << time_span.count() << " seconds";
+ }
+
else {
ss << "unrecognized command! " << cmd;
r = -EINVAL;
}
-bool OSD::ms_verify_authorizer(Connection *con, int peer_type,
- int protocol, bufferlist& authorizer_data, bufferlist& authorizer_reply,
- bool& isvalid, CryptoKey& session_key)
+bool OSD::ms_verify_authorizer(
+ Connection *con, int peer_type,
+ int protocol, bufferlist& authorizer_data, bufferlist& authorizer_reply,
+ bool& isvalid, CryptoKey& session_key,
+ std::unique_ptr<AuthAuthorizerChallenge> *challenge)
{
AuthAuthorizeHandler *authorize_handler = 0;
switch (peer_type) {
uint64_t global_id;
uint64_t auid = CEPH_AUTH_UID_DEFAULT;
- isvalid = authorize_handler->verify_authorizer(
- cct, monc->rotating_secrets.get(),
- authorizer_data, authorizer_reply, name, global_id, caps_info, session_key,
- &auid);
+ RotatingKeyRing *keys = monc->rotating_secrets.get();
+ if (keys) {
+ isvalid = authorize_handler->verify_authorizer(
+ cct, keys,
+ authorizer_data, authorizer_reply, name, global_id, caps_info, session_key,
+ &auid, challenge);
+ } else {
+ dout(10) << __func__ << " no rotating_keys (yet), denied" << dendl;
+ isvalid = false;
+ }
if (isvalid) {
Session *s = static_cast<Session *>(con->get_priv());
handle_scrub(static_cast<MOSDScrub*>(m));
break;
+ case MSG_OSD_FORCE_RECOVERY:
+ handle_force_recovery(m);
+ break;
+
// -- need OSDMap --
case MSG_OSD_PG_CREATE:
struct tm bdt;
time_t tt = now.sec();
localtime_r(&tt, &bdt);
+
+ bool day_permit = false;
+ if (cct->_conf->osd_scrub_begin_week_day < cct->_conf->osd_scrub_end_week_day) {
+ if (bdt.tm_wday >= cct->_conf->osd_scrub_begin_week_day && bdt.tm_wday < cct->_conf->osd_scrub_end_week_day) {
+ day_permit = true;
+ }
+ } else {
+ if (bdt.tm_wday >= cct->_conf->osd_scrub_begin_week_day || bdt.tm_wday < cct->_conf->osd_scrub_end_week_day) {
+ day_permit = true;
+ }
+ }
+
+ if (!day_permit) {
+ dout(20) << __func__ << " should run between week day " << cct->_conf->osd_scrub_begin_week_day
+ << " - " << cct->_conf->osd_scrub_end_week_day
+ << " now " << bdt.tm_wday << " = no" << dendl;
+ return false;
+ }
+
bool time_permit = false;
if (cct->_conf->osd_scrub_begin_hour < cct->_conf->osd_scrub_end_hour) {
if (bdt.tm_hour >= cct->_conf->osd_scrub_begin_hour && bdt.tm_hour < cct->_conf->osd_scrub_end_hour) {
if (!service.can_inc_scrubs_pending()) {
return;
}
+ if (!cct->_conf->osd_scrub_during_recovery && service.is_recovery_active()) {
+ dout(20) << __func__ << " not scheduling scrubs due to active recovery" << dendl;
+ return;
+ }
+
utime_t now = ceph_clock_now();
bool time_permit = scrub_time_permit(now);
break;
}
- if (!cct->_conf->osd_scrub_during_recovery && service.is_recovery_active()) {
- dout(10) << __func__ << "not scheduling scrub of " << scrub.pgid << " due to active recovery ops" << dendl;
- break;
- }
-
if ((scrub.deadline >= now) && !(time_permit && load_is_low)) {
dout(10) << __func__ << " not scheduling scrub for " << scrub.pgid << " due to "
<< (!time_permit ? "time not permit" : "high load") << dendl;
+vector<OSDHealthMetric> OSD::get_health_metrics()
+{
+ vector<OSDHealthMetric> metrics;
+ lock_guard<mutex> pending_creates_locker{pending_creates_lock};
+ auto n_primaries = pending_creates_from_mon;
+ for (const auto& create : pending_creates_from_osd) {
+ if (create.second) {
+ n_primaries++;
+ }
+ }
+ metrics.emplace_back(osd_metric::PENDING_CREATING_PGS, n_primaries);
+ return metrics;
+}
+
// =====================================================
// MAP
void OSD::osdmap_subscribe(version_t epoch, bool force_request)
{
- OSDMapRef osdmap = service.get_osdmap();
- if (osdmap->get_epoch() >= epoch)
+ Mutex::Locker l(osdmap_subscribe_lock);
+ if (latest_subscribed_epoch >= epoch && !force_request)
return;
+ latest_subscribed_epoch = MAX(epoch, latest_subscribed_epoch);
+
if (monc->sub_want_increment("osdmap", epoch, CEPH_SUBSCRIBE_ONETIME) ||
force_request) {
monc->renew_subs();
if (num > 0) {
service.publish_superblock(superblock);
write_superblock(t);
- store->queue_transaction(service.meta_osr.get(), std::move(t), nullptr);
+ int tr = store->queue_transaction(service.meta_osr.get(), std::move(t), nullptr);
+ assert(tr == 0);
}
// we should not remove the cached maps
assert(min <= service.map_cache.cached_key_lower_bound());
dout(10) << __func__ << " require_osd_release reached luminous in "
<< newmap->get_epoch() << dendl;
clear_pg_stat_queue();
- outstanding_pg_stats.clear();
+ clear_outstanding_pg_stats();
}
osdmap = newmap;
if (service.is_preparing_to_stop() || service.is_stopping()) {
service.got_stop_ack();
} else {
- clog->warn() << "map e" << osdmap->get_epoch()
- << " wrongly marked me down at e"
- << osdmap->get_down_at(whoami);
+ clog->warn() << "Monitor daemon marked osd." << whoami << " down, "
+ "but it is still running";
+ clog->debug() << "map e" << osdmap->get_epoch()
+ << " wrongly marked me down at e"
+ << osdmap->get_down_at(whoami);
}
} else if (!osdmap->get_addr(whoami).probably_equals(
client_messenger->get_myaddr())) {
} else if (!osdmap->get_hb_back_addr(whoami).probably_equals(
hb_back_server_messenger->get_myaddr())) {
clog->error() << "map e" << osdmap->get_epoch()
- << " had wrong hb back addr ("
+ << " had wrong heartbeat back addr ("
<< osdmap->get_hb_back_addr(whoami)
<< " != my " << hb_back_server_messenger->get_myaddr()
<< ")";
!osdmap->get_hb_front_addr(whoami).probably_equals(
hb_front_server_messenger->get_myaddr())) {
clog->error() << "map e" << osdmap->get_epoch()
- << " had wrong hb front addr ("
+ << " had wrong heartbeat front addr ("
<< osdmap->get_hb_front_addr(whoami)
<< " != my " << hb_front_server_messenger->get_myaddr()
<< ")";
assert(osd_lock.is_locked());
dout(7) << "consume_map version " << osdmap->get_epoch() << dendl;
+ /** make sure the cluster is speaking in SORTBITWISE, because we don't
+ * speak the older sorting version any more. Be careful not to force
+ * a shutdown if we are merely processing old maps, though.
+ */
+ if (!osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE) && is_active()) {
+ derr << __func__ << " SORTBITWISE flag is not set" << dendl;
+ ceph_abort();
+ }
+
int num_pg_primary = 0, num_pg_replica = 0, num_pg_stray = 0;
list<PGRef> to_remove;
pg->unlock();
}
+
+ lock_guard<mutex> pending_creates_locker{pending_creates_lock};
+ for (auto pg = pending_creates_from_osd.cbegin();
+ pg != pending_creates_from_osd.cend();) {
+ if (osdmap->get_pg_acting_rank(pg->first, whoami) < 0) {
+ pg = pending_creates_from_osd.erase(pg);
+ } else {
+ ++pg;
+ }
+ }
}
for (list<PGRef>::iterator i = to_remove.begin();
logger->set(l_osd_pg_primary, num_pg_primary);
logger->set(l_osd_pg_replica, num_pg_replica);
logger->set(l_osd_pg_stray, num_pg_stray);
+ logger->set(l_osd_pg_removing, remove_wq.get_remove_queue_len());
}
void OSD::activate_map()
dout(7) << "activate_map version " << osdmap->get_epoch() << dendl;
- if (!osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE)) {
- derr << __func__ << " SORTBITWISE flag is not set" << dendl;
- ceph_abort();
- }
-
if (osdmap->test_flag(CEPH_OSDMAP_FULL)) {
dout(10) << " osdmap flagged full, doing onetime osdmap subscribe" << dendl;
osdmap_subscribe(osdmap->get_epoch() + 1, false);
<< dendl;
continue;
}
-
if (handle_pg_peering_evt(
pgid,
history,
service.send_pg_created(pgid.pgid);
}
}
- last_pg_create_epoch = m->epoch;
+ {
+ lock_guard<mutex> pending_creates_locker{pending_creates_lock};
+ if (pending_creates_from_mon == 0) {
+ last_pg_create_epoch = m->epoch;
+ }
+ }
maybe_update_heartbeat_peers();
}
continue;
}
service.share_map_peer(it->first, con.get(), curmap);
- dout(7) << __func__ << " osd " << it->first
+ dout(7) << __func__ << " osd." << it->first
<< " on " << it->second.size() << " PGs" << dendl;
MOSDPGNotify *m = new MOSDPGNotify(curmap->get_epoch(),
it->second);
m->query_epoch,
PG::RemoteBackfillReserved()));
} else if (m->type == MBackfillReserve::REJECT) {
+ // NOTE: this is replica -> primary "i reject your request"
+ // and also primary -> replica "cancel my previously-granted request"
evt = PG::CephPeeringEvtRef(
new PG::CephPeeringEvt(
m->query_epoch,
pg->unlock();
}
+void OSD::handle_force_recovery(Message *m)
+{
+ MOSDForceRecovery *msg = static_cast<MOSDForceRecovery*>(m);
+ assert(msg->get_type() == MSG_OSD_FORCE_RECOVERY);
+
+ vector<PGRef> local_pgs;
+ local_pgs.reserve(msg->forced_pgs.size());
+
+ {
+ RWLock::RLocker l(pg_map_lock);
+ for (auto& i : msg->forced_pgs) {
+ spg_t locpg;
+ if (osdmap->get_primary_shard(i, &locpg)) {
+ auto pg_map_entry = pg_map.find(locpg);
+ if (pg_map_entry != pg_map.end()) {
+ local_pgs.push_back(pg_map_entry->second);
+ }
+ }
+ }
+ }
+
+ if (local_pgs.size()) {
+ service.adjust_pg_priorities(local_pgs, msg->options);
+ }
+
+ msg->put();
+}
/** PGQuery
* from primary to replica | stray
pg->put("PGMap"); // since we've taken it out of map
}
-
// =========================================================
// RECOVERY
return true;
}
+
+void OSDService::adjust_pg_priorities(const vector<PGRef>& pgs, int newflags)
+{
+ if (!pgs.size() || !(newflags & (OFR_BACKFILL | OFR_RECOVERY)))
+ return;
+ int newstate = 0;
+
+ if (newflags & OFR_BACKFILL) {
+ newstate = PG_STATE_FORCED_BACKFILL;
+ } else if (newflags & OFR_RECOVERY) {
+ newstate = PG_STATE_FORCED_RECOVERY;
+ }
+
+ // debug output here may get large, don't generate it if debug level is below
+ // 10 and use abbreviated pg ids otherwise
+ if ((cct)->_conf->subsys.should_gather(ceph_subsys_osd, 10)) {
+ stringstream ss;
+
+ for (auto& i : pgs) {
+ ss << i->get_pgid() << " ";
+ }
+
+ dout(10) << __func__ << " working on " << ss.str() << dendl;
+ }
+
+ if (newflags & OFR_CANCEL) {
+ for (auto& i : pgs) {
+ i->lock();
+ i->_change_recovery_force_mode(newstate, true);
+ i->unlock();
+ }
+ } else {
+ for (auto& i : pgs) {
+ // make sure the PG is in correct state before forcing backfill or recovery, or
+ // else we'll make PG keeping FORCE_* flag forever, requiring osds restart
+ // or forcing somehow recovery/backfill.
+ i->lock();
+ int pgstate = i->get_state();
+ if ( ((newstate == PG_STATE_FORCED_RECOVERY) && (pgstate & (PG_STATE_DEGRADED | PG_STATE_RECOVERY_WAIT | PG_STATE_RECOVERING))) ||
+ ((newstate == PG_STATE_FORCED_BACKFILL) && (pgstate & (PG_STATE_DEGRADED | PG_STATE_BACKFILL_WAIT | PG_STATE_BACKFILLING))) )
+ i->_change_recovery_force_mode(newstate, false);
+ i->unlock();
+ }
+ }
+}
+
void OSD::do_recovery(
PG *pg, epoch_t queued, uint64_t reserved_pushes,
ThreadPool::TPHandle &handle)
* recovery_requeue_callback event, which re-queues the recovery op using
* queue_recovery_after_sleep.
*/
- if (cct->_conf->osd_recovery_sleep > 0 && service.recovery_needs_sleep) {
- PGRef pgref(pg);
- auto recovery_requeue_callback = new FunctionContext([this, pgref, queued, reserved_pushes](int r) {
- dout(20) << "do_recovery wake up at "
- << ceph_clock_now()
- << ", re-queuing recovery" << dendl;
- service.recovery_needs_sleep = false;
- service.queue_recovery_after_sleep(pgref.get(), queued, reserved_pushes);
- });
+ float recovery_sleep = get_osd_recovery_sleep();
+ {
Mutex::Locker l(service.recovery_sleep_lock);
-
- // This is true for the first recovery op and when the previous recovery op
- // has been scheduled in the past. The next recovery op is scheduled after
- // completing the sleep from now.
- if (service.recovery_schedule_time < ceph_clock_now()) {
- service.recovery_schedule_time = ceph_clock_now();
- }
- service.recovery_schedule_time += cct->_conf->osd_recovery_sleep;
- service.recovery_sleep_timer.add_event_at(service.recovery_schedule_time,
- recovery_requeue_callback);
- dout(20) << "Recovery event scheduled at "
- << service.recovery_schedule_time << dendl;
- return;
+ if (recovery_sleep > 0 && service.recovery_needs_sleep) {
+ PGRef pgref(pg);
+ auto recovery_requeue_callback = new FunctionContext([this, pgref, queued, reserved_pushes](int r) {
+ dout(20) << "do_recovery wake up at "
+ << ceph_clock_now()
+ << ", re-queuing recovery" << dendl;
+ Mutex::Locker l(service.recovery_sleep_lock);
+ service.recovery_needs_sleep = false;
+ service.queue_recovery_after_sleep(pgref.get(), queued, reserved_pushes);
+ });
+
+ // This is true for the first recovery op and when the previous recovery op
+ // has been scheduled in the past. The next recovery op is scheduled after
+ // completing the sleep from now.
+ if (service.recovery_schedule_time < ceph_clock_now()) {
+ service.recovery_schedule_time = ceph_clock_now();
+ }
+ service.recovery_schedule_time += recovery_sleep;
+ service.recovery_sleep_timer.add_event_at(service.recovery_schedule_time,
+ recovery_requeue_callback);
+ dout(20) << "Recovery event scheduled at "
+ << service.recovery_schedule_time << dendl;
+ return;
+ }
}
{
- service.recovery_needs_sleep = true;
+ {
+ Mutex::Locker l(service.recovery_sleep_lock);
+ service.recovery_needs_sleep = true;
+ }
+
if (pg->pg_has_reset_since(queued)) {
goto out;
}
if (!more && pg->have_unfound()) {
pg->discover_all_missing(*rctx.query_map);
if (rctx.query_map->empty()) {
- dout(10) << "do_recovery no luck, giving up on this pg for now" << dendl;
+ string action;
+ if (pg->state_test(PG_STATE_BACKFILLING)) {
+ auto evt = PG::CephPeeringEvtRef(new PG::CephPeeringEvt(
+ queued,
+ queued,
+ PG::DeferBackfill(cct->_conf->osd_recovery_retry_interval)));
+ pg->queue_peering_event(evt);
+ action = "in backfill";
+ } else if (pg->state_test(PG_STATE_RECOVERING)) {
+ auto evt = PG::CephPeeringEvtRef(new PG::CephPeeringEvt(
+ queued,
+ queued,
+ PG::DeferRecovery(cct->_conf->osd_recovery_retry_interval)));
+ pg->queue_peering_event(evt);
+ action = "in recovery";
+ } else {
+ action = "already out of recovery/backfill";
+ }
+ dout(10) << __func__ << ": no luck, giving up on this pg for now (" << action << ")" << dendl;
} else {
- dout(10) << "do_recovery no luck, giving up on this pg for now" << dendl;
+ dout(10) << __func__ << ": no luck, giving up on this pg for now (queue_recovery)" << dendl;
pg->queue_recovery();
}
}
bool OSDService::is_recovery_active()
{
- if (recovery_ops_active > 0)
- return true;
-
- return false;
+ return local_reserver.has_reservation() || remote_reserver.has_reservation();
}
// =========================================================
op->osd_trace.keyval("priority", op->get_req()->get_priority());
op->osd_trace.keyval("cost", op->get_req()->get_cost());
op->mark_queued_for_pg();
+ logger->tinc(l_osd_op_before_queue_op_lat, latency);
op_shardedwq.queue(make_pair(pg, PGQueueable(op, epoch)));
}
<< " " << *(op->get_req())
<< " pg " << *pg << dendl;
+ logger->tinc(l_osd_op_before_dequeue_op_lat, latency);
+
Session *session = static_cast<Session *>(
op->get_req()->get_connection()->get_priv());
if (session) {
static const char* KEYS[] = {
"osd_max_backfills",
"osd_min_recovery_priority",
- "osd_op_complaint_time", "osd_op_log_threshold",
- "osd_op_history_size", "osd_op_history_duration",
+ "osd_max_trimming_pgs",
+ "osd_op_complaint_time",
+ "osd_op_log_threshold",
+ "osd_op_history_size",
+ "osd_op_history_duration",
+ "osd_op_history_slow_op_size",
+ "osd_op_history_slow_op_threshold",
"osd_enable_op_tracker",
"osd_map_cache_size",
"osd_map_max_advance",
if (base_pool && base_pool->require_rollback()) {
if ((iter->op.op != CEPH_OSD_OP_READ) &&
(iter->op.op != CEPH_OSD_OP_CHECKSUM) &&
+ (iter->op.op != CEPH_OSD_OP_CMPEXT) &&
(iter->op.op != CEPH_OSD_OP_STAT) &&
(iter->op.op != CEPH_OSD_OP_ISDIRTY) &&
(iter->op.op != CEPH_OSD_OP_UNDIRTY) &&
in_use.insert(out->begin(), out->end());
}
+
// =============================================================
#undef dout_context
uint32_t shard_index = pgid.hash_to_shard(shard_list.size());
auto sdata = shard_list[shard_index];
bool queued = false;
- unsigned pushes_to_free = 0;
{
Mutex::Locker l(sdata->sdata_op_ordering_lock);
auto p = sdata->pg_slots.find(pgid);
++i) {
sdata->_enqueue_front(make_pair(pgid, *i), osd->op_prio_cutoff);
}
- for (auto& q : p->second.to_process) {
- pushes_to_free += q.get_reserved_pushes();
- }
p->second.to_process.clear();
p->second.waiting_for_pg = false;
++p->second.requeue_seq;
queued = true;
}
}
- if (pushes_to_free > 0) {
- osd->service.release_reserved_pushes(pushes_to_free);
- }
if (queued) {
sdata->sdata_lock.Lock();
sdata->sdata_cond.SignalOne();
if (sdata->pqueue->empty()) {
dout(20) << __func__ << " empty q, waiting" << dendl;
// optimistically sleep a moment; maybe another work item will come along.
- sdata->sdata_op_ordering_lock.Unlock();
osd->cct->get_heartbeat_map()->reset_timeout(hb,
osd->cct->_conf->threadpool_default_timeout, 0);
sdata->sdata_lock.Lock();
+ sdata->sdata_op_ordering_lock.Unlock();
sdata->sdata_cond.WaitInterval(sdata->sdata_lock,
utime_t(osd->cct->_conf->threadpool_empty_queue_max_wait, 0));
sdata->sdata_lock.Unlock();
}} // namespace ceph::osd_cmds
+
+std::ostream& operator<<(std::ostream& out, const OSD::io_queue& q) {
+ switch(q) {
+ case OSD::io_queue::prioritized:
+ out << "prioritized";
+ break;
+ case OSD::io_queue::weightedpriority:
+ out << "weightedpriority";
+ break;
+ case OSD::io_queue::mclock_opclass:
+ out << "mclock_opclass";
+ break;
+ case OSD::io_queue::mclock_client:
+ out << "mclock_client";
+ break;
+ }
+ return out;
+}
projects / ceph.git / blobdiff