ceph/src/mgr/DaemonServer.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2016 John Spray <john.spray@redhat.com>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  */
  13
  14 #include "DaemonServer.h"
  15 #include "mgr/Mgr.h"
  16
  17 #include "include/stringify.h"
  18 #include "include/str_list.h"
  19 #include "auth/RotatingKeyRing.h"
  20 #include "json_spirit/json_spirit_writer.h"
  21
  22 #include "mgr/mgr_commands.h"
  23 #include "mgr/OSDHealthMetricCollector.h"
  24 #include "mon/MonCommand.h"
  25
  26 #include "messages/MMgrOpen.h"
  27 #include "messages/MMgrConfigure.h"
  28 #include "messages/MMonMgrReport.h"
  29 #include "messages/MCommand.h"
  30 #include "messages/MCommandReply.h"
  31 #include "messages/MPGStats.h"
  32 #include "messages/MOSDScrub.h"
  33 #include "messages/MOSDForceRecovery.h"
  34 #include "common/errno.h"
  35
  36 #define dout_context g_ceph_context
  37 #define dout_subsys ceph_subsys_mgr
  38 #undef dout_prefix
  39 #define dout_prefix *_dout << "mgr.server " << __func__ << " "
  40
  41
  42
  43 DaemonServer::DaemonServer(MonClient *monc_,
  44                            Finisher &finisher_,
  45                            DaemonStateIndex &daemon_state_,
  46                            ClusterState &cluster_state_,
  47                            PyModuleRegistry &py_modules_,
  48                            LogChannelRef clog_,
  49                            LogChannelRef audit_clog_)
  50     : Dispatcher(g_ceph_context),
  51       client_byte_throttler(new Throttle(g_ceph_context, "mgr_client_bytes",
  52                                          g_conf->get_val<uint64_t>("mgr_client_bytes"))),
  53       client_msg_throttler(new Throttle(g_ceph_context, "mgr_client_messages",
  54                                         g_conf->get_val<uint64_t>("mgr_client_messages"))),
  55       osd_byte_throttler(new Throttle(g_ceph_context, "mgr_osd_bytes",
  56                                       g_conf->get_val<uint64_t>("mgr_osd_bytes"))),
  57       osd_msg_throttler(new Throttle(g_ceph_context, "mgr_osd_messsages",
  58                                      g_conf->get_val<uint64_t>("mgr_osd_messages"))),
  59       mds_byte_throttler(new Throttle(g_ceph_context, "mgr_mds_bytes",
  60                                       g_conf->get_val<uint64_t>("mgr_mds_bytes"))),
  61       mds_msg_throttler(new Throttle(g_ceph_context, "mgr_mds_messsages",
  62                                      g_conf->get_val<uint64_t>("mgr_mds_messages"))),
  63       mon_byte_throttler(new Throttle(g_ceph_context, "mgr_mon_bytes",
  64                                       g_conf->get_val<uint64_t>("mgr_mon_bytes"))),
  65       mon_msg_throttler(new Throttle(g_ceph_context, "mgr_mon_messsages",
  66                                      g_conf->get_val<uint64_t>("mgr_mon_messages"))),
  67       msgr(nullptr),
  68       monc(monc_),
  69       finisher(finisher_),
  70       daemon_state(daemon_state_),
  71       cluster_state(cluster_state_),
  72       py_modules(py_modules_),
  73       clog(clog_),
  74       audit_clog(audit_clog_),
  75       auth_cluster_registry(g_ceph_context,
  76                     g_conf->auth_supported.empty() ?
  77                       g_conf->auth_cluster_required :
  78                       g_conf->auth_supported),
  79       auth_service_registry(g_ceph_context,
  80                    g_conf->auth_supported.empty() ?
  81                       g_conf->auth_service_required :
  82                       g_conf->auth_supported),
  83       lock("DaemonServer"),
  84       pgmap_ready(false)
  85 {
  86   g_conf->add_observer(this);
  87 }
  88
  89 DaemonServer::~DaemonServer() {
  90   delete msgr;
  91   g_conf->remove_observer(this);
  92 }
  93
  94 int DaemonServer::init(uint64_t gid, entity_addr_t client_addr)
  95 {
  96   // Initialize Messenger
  97   std::string public_msgr_type = g_conf->ms_public_type.empty() ?
  98     g_conf->get_val<std::string>("ms_type") : g_conf->ms_public_type;
  99   msgr = Messenger::create(g_ceph_context, public_msgr_type,
 100                            entity_name_t::MGR(gid),
 101                            "mgr",
 102                            getpid(), 0);
 103   msgr->set_default_policy(Messenger::Policy::stateless_server(0));
 104
 105   // throttle clients
 106   msgr->set_policy_throttlers(entity_name_t::TYPE_CLIENT,
 107                               client_byte_throttler.get(),
 108                               client_msg_throttler.get());
 109
 110   // servers
 111   msgr->set_policy_throttlers(entity_name_t::TYPE_OSD,
 112                               osd_byte_throttler.get(),
 113                               osd_msg_throttler.get());
 114   msgr->set_policy_throttlers(entity_name_t::TYPE_MDS,
 115                               mds_byte_throttler.get(),
 116                               mds_msg_throttler.get());
 117   msgr->set_policy_throttlers(entity_name_t::TYPE_MON,
 118                               mon_byte_throttler.get(),
 119                               mon_msg_throttler.get());
 120
 121   int r = msgr->bind(g_conf->public_addr);
 122   if (r < 0) {
 123     derr << "unable to bind mgr to " << g_conf->public_addr << dendl;
 124     return r;
 125   }
 126
 127   msgr->set_myname(entity_name_t::MGR(gid));
 128   msgr->set_addr_unknowns(client_addr);
 129
 130   msgr->start();
 131   msgr->add_dispatcher_tail(this);
 132
 133   started_at = ceph_clock_now();
 134
 135   return 0;
 136 }
 137
 138 entity_addr_t DaemonServer::get_myaddr() const
 139 {
 140   return msgr->get_myaddr();
 141 }
 142
 143
 144 bool DaemonServer::ms_verify_authorizer(
 145   Connection *con,
 146   int peer_type,
 147   int protocol,
 148   ceph::bufferlist& authorizer_data,
 149   ceph::bufferlist& authorizer_reply,
 150   bool& is_valid,
 151   CryptoKey& session_key,
 152   std::unique_ptr<AuthAuthorizerChallenge> *challenge)
 153 {
 154   AuthAuthorizeHandler *handler = nullptr;
 155   if (peer_type == CEPH_ENTITY_TYPE_OSD ||
 156       peer_type == CEPH_ENTITY_TYPE_MON ||
 157       peer_type == CEPH_ENTITY_TYPE_MDS ||
 158       peer_type == CEPH_ENTITY_TYPE_MGR) {
 159     handler = auth_cluster_registry.get_handler(protocol);
 160   } else {
 161     handler = auth_service_registry.get_handler(protocol);
 162   }
 163   if (!handler) {
 164     dout(0) << "No AuthAuthorizeHandler found for protocol " << protocol << dendl;
 165     is_valid = false;
 166     return true;
 167   }
 168
 169   MgrSessionRef s(new MgrSession(cct));
 170   s->inst.addr = con->get_peer_addr();
 171   AuthCapsInfo caps_info;
 172
 173   RotatingKeyRing *keys = monc->rotating_secrets.get();
 174   if (keys) {
 175     is_valid = handler->verify_authorizer(
 176       cct, keys,
 177       authorizer_data,
 178       authorizer_reply, s->entity_name,
 179       s->global_id, caps_info,
 180       session_key,
 181       nullptr,
 182       challenge);
 183   } else {
 184     dout(10) << __func__ << " no rotating_keys (yet), denied" << dendl;
 185     is_valid = false;
 186   }
 187
 188   if (is_valid) {
 189     if (caps_info.allow_all) {
 190       dout(10) << " session " << s << " " << s->entity_name
 191                << " allow_all" << dendl;
 192       s->caps.set_allow_all();
 193     }
 194     if (caps_info.caps.length() > 0) {
 195       bufferlist::iterator p = caps_info.caps.begin();
 196       string str;
 197       try {
 198         ::decode(str, p);
 199       }
 200       catch (buffer::error& e) {
 201       }
 202       bool success = s->caps.parse(str);
 203       if (success) {
 204         dout(10) << " session " << s << " " << s->entity_name
 205                  << " has caps " << s->caps << " '" << str << "'" << dendl;
 206       } else {
 207         dout(10) << " session " << s << " " << s->entity_name
 208                  << " failed to parse caps '" << str << "'" << dendl;
 209         is_valid = false;
 210       }
 211     }
 212     con->set_priv(s->get());
 213
 214     if (peer_type == CEPH_ENTITY_TYPE_OSD) {
 215       Mutex::Locker l(lock);
 216       s->osd_id = atoi(s->entity_name.get_id().c_str());
 217       dout(10) << "registering osd." << s->osd_id << " session "
 218                << s << " con " << con << dendl;
 219       osd_cons[s->osd_id].insert(con);
 220     }
 221   }
 222
 223   return true;
 224 }
 225
 226
 227 bool DaemonServer::ms_get_authorizer(int dest_type,
 228     AuthAuthorizer **authorizer, bool force_new)
 229 {
 230   dout(10) << "type=" << ceph_entity_type_name(dest_type) << dendl;
 231
 232   if (dest_type == CEPH_ENTITY_TYPE_MON) {
 233     return true;
 234   }
 235
 236   if (force_new) {
 237     if (monc->wait_auth_rotating(10) < 0)
 238       return false;
 239   }
 240
 241   *authorizer = monc->build_authorizer(dest_type);
 242   dout(20) << "got authorizer " << *authorizer << dendl;
 243   return *authorizer != NULL;
 244 }
 245
 246 bool DaemonServer::ms_handle_reset(Connection *con)
 247 {
 248   if (con->get_peer_type() == CEPH_ENTITY_TYPE_OSD) {
 249     MgrSessionRef session(static_cast<MgrSession*>(con->get_priv()));
 250     if (!session) {
 251       return false;
 252     }
 253     session->put(); // SessionRef takes a ref
 254     Mutex::Locker l(lock);
 255     dout(10) << "unregistering osd." << session->osd_id
 256              << "  session " << session << " con " << con << dendl;
 257     osd_cons[session->osd_id].erase(con);
 258
 259     auto iter = daemon_connections.find(con);
 260     if (iter != daemon_connections.end()) {
 261       daemon_connections.erase(iter);
 262     }
 263   }
 264   return false;
 265 }
 266
 267 bool DaemonServer::ms_handle_refused(Connection *con)
 268 {
 269   // do nothing for now
 270   return false;
 271 }
 272
 273 bool DaemonServer::ms_dispatch(Message *m)
 274 {
 275   // Note that we do *not* take ::lock here, in order to avoid
 276   // serializing all message handling.  It's up to each handler
 277   // to take whatever locks it needs.
 278   switch (m->get_type()) {
 279     case MSG_PGSTATS:
 280       cluster_state.ingest_pgstats(static_cast<MPGStats*>(m));
 281       maybe_ready(m->get_source().num());
 282       m->put();
 283       return true;
 284     case MSG_MGR_REPORT:
 285       return handle_report(static_cast<MMgrReport*>(m));
 286     case MSG_MGR_OPEN:
 287       return handle_open(static_cast<MMgrOpen*>(m));
 288     case MSG_COMMAND:
 289       return handle_command(static_cast<MCommand*>(m));
 290     default:
 291       dout(1) << "Unhandled message type " << m->get_type() << dendl;
 292       return false;
 293   };
 294 }
 295
 296 void DaemonServer::maybe_ready(int32_t osd_id)
 297 {
 298   if (pgmap_ready.load()) {
 299     // Fast path: we don't need to take lock because pgmap_ready
 300     // is already set
 301   } else {
 302     Mutex::Locker l(lock);
 303
 304     if (reported_osds.find(osd_id) == reported_osds.end()) {
 305       dout(4) << "initial report from osd " << osd_id << dendl;
 306       reported_osds.insert(osd_id);
 307       std::set<int32_t> up_osds;
 308
 309       cluster_state.with_osdmap([&](const OSDMap& osdmap) {
 310           osdmap.get_up_osds(up_osds);
 311       });
 312
 313       std::set<int32_t> unreported_osds;
 314       std::set_difference(up_osds.begin(), up_osds.end(),
 315                           reported_osds.begin(), reported_osds.end(),
 316                           std::inserter(unreported_osds, unreported_osds.begin()));
 317
 318       if (unreported_osds.size() == 0) {
 319         dout(4) << "all osds have reported, sending PG state to mon" << dendl;
 320         pgmap_ready = true;
 321         reported_osds.clear();
 322         // Avoid waiting for next tick
 323         send_report();
 324       } else {
 325         dout(4) << "still waiting for " << unreported_osds.size() << " osds"
 326                    " to report in before PGMap is ready" << dendl;
 327       }
 328     }
 329   }
 330 }
 331
 332 void DaemonServer::shutdown()
 333 {
 334   dout(10) << "begin" << dendl;
 335   msgr->shutdown();
 336   msgr->wait();
 337   dout(10) << "done" << dendl;
 338 }
 339
 340
 341
 342 bool DaemonServer::handle_open(MMgrOpen *m)
 343 {
 344   Mutex::Locker l(lock);
 345
 346   DaemonKey key;
 347   if (!m->service_name.empty()) {
 348     key.first = m->service_name;
 349   } else {
 350     key.first = ceph_entity_type_name(m->get_connection()->get_peer_type());
 351   }
 352   key.second = m->daemon_name;
 353
 354   dout(4) << "from " << m->get_connection() << "  " << key << dendl;
 355
 356   _send_configure(m->get_connection());
 357
 358   DaemonStatePtr daemon;
 359   if (daemon_state.exists(key)) {
 360     daemon = daemon_state.get(key);
 361   }
 362   if (daemon) {
 363     dout(20) << "updating existing DaemonState for " << m->daemon_name << dendl;
 364     Mutex::Locker l(daemon->lock);
 365     daemon->perf_counters.clear();
 366   }
 367
 368   if (m->service_daemon) {
 369     if (!daemon) {
 370       dout(4) << "constructing new DaemonState for " << key << dendl;
 371       daemon = std::make_shared<DaemonState>(daemon_state.types);
 372       daemon->key = key;
 373       if (m->daemon_metadata.count("hostname")) {
 374         daemon->hostname = m->daemon_metadata["hostname"];
 375       }
 376       daemon_state.insert(daemon);
 377     }
 378     Mutex::Locker l(daemon->lock);
 379     daemon->service_daemon = true;
 380     daemon->metadata = m->daemon_metadata;
 381     daemon->service_status = m->daemon_status;
 382
 383     utime_t now = ceph_clock_now();
 384     auto d = pending_service_map.get_daemon(m->service_name,
 385                                             m->daemon_name);
 386     if (d->gid != (uint64_t)m->get_source().num()) {
 387       dout(10) << "registering " << key << " in pending_service_map" << dendl;
 388       d->gid = m->get_source().num();
 389       d->addr = m->get_source_addr();
 390       d->start_epoch = pending_service_map.epoch;
 391       d->start_stamp = now;
 392       d->metadata = m->daemon_metadata;
 393       pending_service_map_dirty = pending_service_map.epoch;
 394     }
 395   }
 396
 397   if (m->get_connection()->get_peer_type() != entity_name_t::TYPE_CLIENT &&
 398       m->service_name.empty())
 399   {
 400     // Store in set of the daemon/service connections, i.e. those
 401     // connections that require an update in the event of stats
 402     // configuration changes.
 403     daemon_connections.insert(m->get_connection());
 404   }
 405
 406   m->put();
 407   return true;
 408 }
 409
 410 bool DaemonServer::handle_report(MMgrReport *m)
 411 {
 412   DaemonKey key;
 413   if (!m->service_name.empty()) {
 414     key.first = m->service_name;
 415   } else {
 416     key.first = ceph_entity_type_name(m->get_connection()->get_peer_type());
 417   }
 418   key.second = m->daemon_name;
 419
 420   dout(4) << "from " << m->get_connection() << " " << key << dendl;
 421
 422   if (m->get_connection()->get_peer_type() == entity_name_t::TYPE_CLIENT &&
 423       m->service_name.empty()) {
 424     // Clients should not be sending us stats unless they are declaring
 425     // themselves to be a daemon for some service.
 426     dout(4) << "rejecting report from non-daemon client " << m->daemon_name
 427             << dendl;
 428     m->get_connection()->mark_down();
 429     m->put();
 430     return true;
 431   }
 432
 433   // Look up the DaemonState
 434   DaemonStatePtr daemon;
 435   if (daemon_state.exists(key)) {
 436     dout(20) << "updating existing DaemonState for " << key << dendl;
 437     daemon = daemon_state.get(key);
 438   } else {
 439     // we don't know the hostname at this stage, reject MMgrReport here.
 440     dout(5) << "rejecting report from " << key << ", since we do not have its metadata now."
 441             << dendl;
 442
 443     // issue metadata request in background
 444     if (!daemon_state.is_updating(key) &&
 445         (key.first == "osd" || key.first == "mds")) {
 446
 447       std::ostringstream oss;
 448       auto c = new MetadataUpdate(daemon_state, key);
 449       if (key.first == "osd") {
 450         oss << "{\"prefix\": \"osd metadata\", \"id\": "
 451             << key.second<< "}";
 452
 453       } else if (key.first == "mds") {
 454         c->set_default("addr", stringify(m->get_source_addr()));
 455         oss << "{\"prefix\": \"mds metadata\", \"who\": \""
 456             << key.second << "\"}";
 457
 458       } else {
 459         ceph_abort();
 460       }
 461
 462       monc->start_mon_command({oss.str()}, {}, &c->outbl, &c->outs, c);
 463     }
 464
 465     {
 466       Mutex::Locker l(lock);
 467       // kill session
 468       MgrSessionRef session(static_cast<MgrSession*>(m->get_connection()->get_priv()));
 469       if (!session) {
 470         return false;
 471       }
 472       m->get_connection()->mark_down();
 473       session->put();
 474
 475       dout(10) << "unregistering osd." << session->osd_id
 476                << "  session " << session << " con " << m->get_connection() << dendl;
 477
 478       if (osd_cons.find(session->osd_id) != osd_cons.end()) {
 479            osd_cons[session->osd_id].erase(m->get_connection());
 480       }
 481
 482       auto iter = daemon_connections.find(m->get_connection());
 483       if (iter != daemon_connections.end()) {
 484         daemon_connections.erase(iter);
 485       }
 486     }
 487
 488     return false;
 489   }
 490
 491   // Update the DaemonState
 492   assert(daemon != nullptr);
 493   {
 494     Mutex::Locker l(daemon->lock);
 495     auto &daemon_counters = daemon->perf_counters;
 496     daemon_counters.update(m);
 497
 498     if (daemon->service_daemon) {
 499       utime_t now = ceph_clock_now();
 500       if (m->daemon_status) {
 501         daemon->service_status = *m->daemon_status;
 502         daemon->service_status_stamp = now;
 503       }
 504       daemon->last_service_beacon = now;
 505     } else if (m->daemon_status) {
 506       derr << "got status from non-daemon " << key << dendl;
 507     }
 508     if (m->get_connection()->peer_is_osd()) {
 509       // only OSD sends health_checks to me now
 510       daemon->osd_health_metrics = std::move(m->osd_health_metrics);
 511     }
 512   }
 513
 514   // if there are any schema updates, notify the python modules
 515   if (!m->declare_types.empty() || !m->undeclare_types.empty()) {
 516     ostringstream oss;
 517     oss << key.first << '.' << key.second;
 518     py_modules.notify_all("perf_schema_update", oss.str());
 519   }
 520
 521   m->put();
 522   return true;
 523 }
 524
 525
 526 void DaemonServer::_generate_command_map(
 527   map<string,cmd_vartype>& cmdmap,
 528   map<string,string> &param_str_map)
 529 {
 530   for (map<string,cmd_vartype>::const_iterator p = cmdmap.begin();
 531        p != cmdmap.end(); ++p) {
 532     if (p->first == "prefix")
 533       continue;
 534     if (p->first == "caps") {
 535       vector<string> cv;
 536       if (cmd_getval(g_ceph_context, cmdmap, "caps", cv) &&
 537           cv.size() % 2 == 0) {
 538         for (unsigned i = 0; i < cv.size(); i += 2) {
 539           string k = string("caps_") + cv[i];
 540           param_str_map[k] = cv[i + 1];
 541         }
 542         continue;
 543       }
 544     }
 545     param_str_map[p->first] = cmd_vartype_stringify(p->second);
 546   }
 547 }
 548
 549 const MonCommand *DaemonServer::_get_mgrcommand(
 550   const string &cmd_prefix,
 551   const std::vector<MonCommand> &cmds)
 552 {
 553   const MonCommand *this_cmd = nullptr;
 554   for (const auto &cmd : cmds) {
 555     if (cmd.cmdstring.compare(0, cmd_prefix.size(), cmd_prefix) == 0) {
 556       this_cmd = &cmd;
 557       break;
 558     }
 559   }
 560   return this_cmd;
 561 }
 562
 563 bool DaemonServer::_allowed_command(
 564   MgrSession *s,
 565   const string &module,
 566   const string &prefix,
 567   const map<string,cmd_vartype>& cmdmap,
 568   const map<string,string>& param_str_map,
 569   const MonCommand *this_cmd) {
 570
 571   if (s->entity_name.is_mon()) {
 572     // mon is all-powerful.  even when it is forwarding commands on behalf of
 573     // old clients; we expect the mon is validating commands before proxying!
 574     return true;
 575   }
 576
 577   bool cmd_r = this_cmd->requires_perm('r');
 578   bool cmd_w = this_cmd->requires_perm('w');
 579   bool cmd_x = this_cmd->requires_perm('x');
 580
 581   bool capable = s->caps.is_capable(
 582     g_ceph_context,
 583     CEPH_ENTITY_TYPE_MGR,
 584     s->entity_name,
 585     module, prefix, param_str_map,
 586     cmd_r, cmd_w, cmd_x);
 587
 588   dout(10) << " " << s->entity_name << " "
 589            << (capable ? "" : "not ") << "capable" << dendl;
 590   return capable;
 591 }
 592
 593 bool DaemonServer::handle_command(MCommand *m)
 594 {
 595   Mutex::Locker l(lock);
 596   int r = 0;
 597   std::stringstream ss;
 598   std::string prefix;
 599
 600   assert(lock.is_locked_by_me());
 601
 602   /**
 603    * The working data for processing an MCommand.  This lives in
 604    * a class to enable passing it into other threads for processing
 605    * outside of the thread/locks that called handle_command.
 606    */
 607   class CommandContext
 608   {
 609     public:
 610     MCommand *m;
 611     bufferlist odata;
 612     cmdmap_t cmdmap;
 613
 614     CommandContext(MCommand *m_)
 615       : m(m_)
 616     {
 617     }
 618
 619     ~CommandContext()
 620     {
 621       m->put();
 622     }
 623
 624     void reply(int r, const std::stringstream &ss)
 625     {
 626       reply(r, ss.str());
 627     }
 628
 629     void reply(int r, const std::string &rs)
 630     {
 631       // Let the connection drop as soon as we've sent our response
 632       ConnectionRef con = m->get_connection();
 633       if (con) {
 634         con->mark_disposable();
 635       }
 636
 637       dout(1) << "handle_command " << cpp_strerror(r) << " " << rs << dendl;
 638       if (con) {
 639         MCommandReply *reply = new MCommandReply(r, rs);
 640         reply->set_tid(m->get_tid());
 641         reply->set_data(odata);
 642         con->send_message(reply);
 643       }
 644     }
 645   };
 646
 647   /**
 648    * A context for receiving a bufferlist/error string from a background
 649    * function and then calling back to a CommandContext when it's done
 650    */
 651   class ReplyOnFinish : public Context {
 652     std::shared_ptr<CommandContext> cmdctx;
 653
 654   public:
 655     bufferlist from_mon;
 656     string outs;
 657
 658     ReplyOnFinish(std::shared_ptr<CommandContext> cmdctx_)
 659       : cmdctx(cmdctx_)
 660     {}
 661     void finish(int r) override {
 662       cmdctx->odata.claim_append(from_mon);
 663       cmdctx->reply(r, outs);
 664     }
 665   };
 666
 667   std::shared_ptr<CommandContext> cmdctx = std::make_shared<CommandContext>(m);
 668
 669   MgrSessionRef session(static_cast<MgrSession*>(m->get_connection()->get_priv()));
 670   if (!session) {
 671     return true;
 672   }
 673   session->put(); // SessionRef takes a ref
 674   if (session->inst.name == entity_name_t())
 675     session->inst.name = m->get_source();
 676
 677   std::string format;
 678   boost::scoped_ptr<Formatter> f;
 679   map<string,string> param_str_map;
 680
 681   if (!cmdmap_from_json(m->cmd, &(cmdctx->cmdmap), ss)) {
 682     cmdctx->reply(-EINVAL, ss);
 683     return true;
 684   }
 685
 686   {
 687     cmd_getval(g_ceph_context, cmdctx->cmdmap, "format", format, string("plain"));
 688     f.reset(Formatter::create(format));
 689   }
 690
 691   cmd_getval(cct, cmdctx->cmdmap, "prefix", prefix);
 692
 693   dout(4) << "decoded " << cmdctx->cmdmap.size() << dendl;
 694   dout(4) << "prefix=" << prefix << dendl;
 695
 696   if (prefix == "get_command_descriptions") {
 697     dout(10) << "reading commands from python modules" << dendl;
 698     const auto py_commands = py_modules.get_commands();
 699
 700     int cmdnum = 0;
 701     JSONFormatter f;
 702     f.open_object_section("command_descriptions");
 703
 704     auto dump_cmd = [&cmdnum, &f](const MonCommand &mc){
 705       ostringstream secname;
 706       secname << "cmd" << setfill('0') << std::setw(3) << cmdnum;
 707       dump_cmddesc_to_json(&f, secname.str(), mc.cmdstring, mc.helpstring,
 708                            mc.module, mc.req_perms, mc.availability, 0);
 709       cmdnum++;
 710     };
 711
 712     for (const auto &pyc : py_commands) {
 713       dump_cmd(pyc);
 714     }
 715
 716     for (const auto &mgr_cmd : mgr_commands) {
 717       dump_cmd(mgr_cmd);
 718     }
 719
 720     f.close_section();  // command_descriptions
 721     f.flush(cmdctx->odata);
 722     cmdctx->reply(0, ss);
 723     return true;
 724   }
 725
 726   // lookup command
 727   const MonCommand *mgr_cmd = _get_mgrcommand(prefix, mgr_commands);
 728   _generate_command_map(cmdctx->cmdmap, param_str_map);
 729   if (!mgr_cmd) {
 730     MonCommand py_command = {"", "", "py", "rw", "cli"};
 731     if (!_allowed_command(session.get(), py_command.module, prefix, cmdctx->cmdmap,
 732                           param_str_map, &py_command)) {
 733       dout(1) << " access denied" << dendl;
 734       ss << "access denied; does your client key have mgr caps?"
 735         " See http://docs.ceph.com/docs/master/mgr/administrator/#client-authentication";
 736       cmdctx->reply(-EACCES, ss);
 737       return true;
 738     }
 739   } else {
 740     // validate user's permissions for requested command
 741     if (!_allowed_command(session.get(), mgr_cmd->module, prefix, cmdctx->cmdmap,
 742                           param_str_map, mgr_cmd)) {
 743       dout(1) << " access denied" << dendl;
 744       audit_clog->info() << "from='" << session->inst << "' "
 745                          << "entity='" << session->entity_name << "' "
 746                          << "cmd=" << m->cmd << ":  access denied";
 747       ss << "access denied' does your client key have mgr caps?"
 748         " See http://docs.ceph.com/docs/master/mgr/administrator/#client-authentication";
 749       cmdctx->reply(-EACCES, ss);
 750       return true;
 751     }
 752   }
 753
 754   audit_clog->debug()
 755     << "from='" << session->inst << "' "
 756     << "entity='" << session->entity_name << "' "
 757     << "cmd=" << m->cmd << ": dispatch";
 758
 759   // ----------------
 760   // service map commands
 761   if (prefix == "service dump") {
 762     if (!f)
 763       f.reset(Formatter::create("json-pretty"));
 764     cluster_state.with_servicemap([&](const ServiceMap &service_map) {
 765         f->dump_object("service_map", service_map);
 766       });
 767     f->flush(cmdctx->odata);
 768     cmdctx->reply(0, ss);
 769     return true;
 770   }
 771   if (prefix == "service status") {
 772     if (!f)
 773       f.reset(Formatter::create("json-pretty"));
 774     // only include state from services that are in the persisted service map
 775     f->open_object_section("service_status");
 776     for (auto& p : pending_service_map.services) {
 777       f->open_object_section(p.first.c_str());
 778       for (auto& q : p.second.daemons) {
 779         f->open_object_section(q.first.c_str());
 780         DaemonKey key(p.first, q.first);
 781         assert(daemon_state.exists(key));
 782         auto daemon = daemon_state.get(key);
 783         Mutex::Locker l(daemon->lock);
 784         f->dump_stream("status_stamp") << daemon->service_status_stamp;
 785         f->dump_stream("last_beacon") << daemon->last_service_beacon;
 786         f->open_object_section("status");
 787         for (auto& r : daemon->service_status) {
 788           f->dump_string(r.first.c_str(), r.second);
 789         }
 790         f->close_section();
 791         f->close_section();
 792       }
 793       f->close_section();
 794     }
 795     f->close_section();
 796     f->flush(cmdctx->odata);
 797     cmdctx->reply(0, ss);
 798     return true;
 799   }
 800
 801   if (prefix == "config set") {
 802     std::string key;
 803     std::string val;
 804     cmd_getval(cct, cmdctx->cmdmap, "key", key);
 805     cmd_getval(cct, cmdctx->cmdmap, "value", val);
 806     r = cct->_conf->set_val(key, val, true, &ss);
 807     if (r == 0) {
 808       cct->_conf->apply_changes(nullptr);
 809     }
 810     cmdctx->reply(0, ss);
 811     return true;
 812   }
 813
 814   // -----------
 815   // PG commands
 816
 817   if (prefix == "pg scrub" ||
 818       prefix == "pg repair" ||
 819       prefix == "pg deep-scrub") {
 820     string scrubop = prefix.substr(3, string::npos);
 821     pg_t pgid;
 822     string pgidstr;
 823     cmd_getval(g_ceph_context, cmdctx->cmdmap, "pgid", pgidstr);
 824     if (!pgid.parse(pgidstr.c_str())) {
 825       ss << "invalid pgid '" << pgidstr << "'";
 826       cmdctx->reply(-EINVAL, ss);
 827       return true;
 828     }
 829     bool pg_exists = false;
 830     cluster_state.with_osdmap([&](const OSDMap& osdmap) {
 831         pg_exists = osdmap.pg_exists(pgid);
 832       });
 833     if (!pg_exists) {
 834       ss << "pg " << pgid << " dne";
 835       cmdctx->reply(-ENOENT, ss);
 836       return true;
 837     }
 838     int acting_primary = -1;
 839     cluster_state.with_osdmap([&](const OSDMap& osdmap) {
 840         acting_primary = osdmap.get_pg_acting_primary(pgid);
 841       });
 842     if (acting_primary == -1) {
 843       ss << "pg " << pgid << " has no primary osd";
 844       cmdctx->reply(-EAGAIN, ss);
 845       return true;
 846     }
 847     auto p = osd_cons.find(acting_primary);
 848     if (p == osd_cons.end()) {
 849       ss << "pg " << pgid << " primary osd." << acting_primary
 850          << " is not currently connected";
 851       cmdctx->reply(-EAGAIN, ss);
 852       return true;
 853     }
 854     vector<pg_t> pgs = { pgid };
 855     for (auto& con : p->second) {
 856       con->send_message(new MOSDScrub(monc->get_fsid(),
 857                                       pgs,
 858                                       scrubop == "repair",
 859                                       scrubop == "deep-scrub"));
 860     }
 861     ss << "instructing pg " << pgid << " on osd." << acting_primary
 862        << " to " << scrubop;
 863     cmdctx->reply(0, ss);
 864     return true;
 865   } else if (prefix == "osd scrub" ||
 866               prefix == "osd deep-scrub" ||
 867               prefix == "osd repair") {
 868     string whostr;
 869     cmd_getval(g_ceph_context, cmdctx->cmdmap, "who", whostr);
 870     vector<string> pvec;
 871     get_str_vec(prefix, pvec);
 872
 873     set<int> osds;
 874     if (whostr == "*" || whostr == "all" || whostr == "any") {
 875       cluster_state.with_osdmap([&](const OSDMap& osdmap) {
 876           for (int i = 0; i < osdmap.get_max_osd(); i++)
 877             if (osdmap.is_up(i)) {
 878               osds.insert(i);
 879             }
 880         });
 881     } else {
 882       long osd = parse_osd_id(whostr.c_str(), &ss);
 883       if (osd < 0) {
 884         ss << "invalid osd '" << whostr << "'";
 885         cmdctx->reply(-EINVAL, ss);
 886         return true;
 887       }
 888       cluster_state.with_osdmap([&](const OSDMap& osdmap) {
 889           if (osdmap.is_up(osd)) {
 890             osds.insert(osd);
 891           }
 892         });
 893       if (osds.empty()) {
 894         ss << "osd." << osd << " is not up";
 895         cmdctx->reply(-EAGAIN, ss);
 896         return true;
 897       }
 898     }
 899     set<int> sent_osds, failed_osds;
 900     for (auto osd : osds) {
 901       auto p = osd_cons.find(osd);
 902       if (p == osd_cons.end()) {
 903         failed_osds.insert(osd);
 904       } else {
 905         sent_osds.insert(osd);
 906         for (auto& con : p->second) {
 907           con->send_message(new MOSDScrub(monc->get_fsid(),
 908                                           pvec.back() == "repair",
 909                                           pvec.back() == "deep-scrub"));
 910         }
 911       }
 912     }
 913     if (failed_osds.size() == osds.size()) {
 914       ss << "failed to instruct osd(s) " << osds << " to " << pvec.back()
 915          << " (not connected)";
 916       r = -EAGAIN;
 917     } else {
 918       ss << "instructed osd(s) " << sent_osds << " to " << pvec.back();
 919       if (!failed_osds.empty()) {
 920         ss << "; osd(s) " << failed_osds << " were not connected";
 921       }
 922       r = 0;
 923     }
 924     cmdctx->reply(0, ss);
 925     return true;
 926   } else if (prefix == "osd reweight-by-pg" ||
 927              prefix == "osd reweight-by-utilization" ||
 928              prefix == "osd test-reweight-by-pg" ||
 929              prefix == "osd test-reweight-by-utilization") {
 930     bool by_pg =
 931       prefix == "osd reweight-by-pg" || prefix == "osd test-reweight-by-pg";
 932     bool dry_run =
 933       prefix == "osd test-reweight-by-pg" ||
 934       prefix == "osd test-reweight-by-utilization";
 935     int64_t oload;
 936     cmd_getval(g_ceph_context, cmdctx->cmdmap, "oload", oload, int64_t(120));
 937     set<int64_t> pools;
 938     vector<string> poolnames;
 939     cmd_getval(g_ceph_context, cmdctx->cmdmap, "pools", poolnames);
 940     cluster_state.with_osdmap([&](const OSDMap& osdmap) {
 941         for (const auto& poolname : poolnames) {
 942           int64_t pool = osdmap.lookup_pg_pool_name(poolname);
 943           if (pool < 0) {
 944             ss << "pool '" << poolname << "' does not exist";
 945             r = -ENOENT;
 946           }
 947           pools.insert(pool);
 948         }
 949       });
 950     if (r) {
 951       cmdctx->reply(r, ss);
 952       return true;
 953     }
 954     double max_change = g_conf->mon_reweight_max_change;
 955     cmd_getval(g_ceph_context, cmdctx->cmdmap, "max_change", max_change);
 956     if (max_change <= 0.0) {
 957       ss << "max_change " << max_change << " must be positive";
 958       cmdctx->reply(-EINVAL, ss);
 959       return true;
 960     }
 961     int64_t max_osds = g_conf->mon_reweight_max_osds;
 962     cmd_getval(g_ceph_context, cmdctx->cmdmap, "max_osds", max_osds);
 963     if (max_osds <= 0) {
 964       ss << "max_osds " << max_osds << " must be positive";
 965       cmdctx->reply(-EINVAL, ss);
 966       return true;
 967     }
 968     string no_increasing;
 969     cmd_getval(g_ceph_context, cmdctx->cmdmap, "no_increasing", no_increasing);
 970     string out_str;
 971     mempool::osdmap::map<int32_t, uint32_t> new_weights;
 972     r = cluster_state.with_pgmap([&](const PGMap& pgmap) {
 973         return cluster_state.with_osdmap([&](const OSDMap& osdmap) {
 974             return reweight::by_utilization(osdmap, pgmap,
 975                                             oload,
 976                                             max_change,
 977                                             max_osds,
 978                                             by_pg,
 979                                             pools.empty() ? NULL : &pools,
 980                                             no_increasing == "--no-increasing",
 981                                             &new_weights,
 982                                             &ss, &out_str, f.get());
 983           });
 984       });
 985     if (r >= 0) {
 986       dout(10) << "reweight::by_utilization: finished with " << out_str << dendl;
 987     }
 988     if (f) {
 989       f->flush(cmdctx->odata);
 990     } else {
 991       cmdctx->odata.append(out_str);
 992     }
 993     if (r < 0) {
 994       ss << "FAILED reweight-by-pg";
 995       cmdctx->reply(r, ss);
 996       return true;
 997     } else if (r == 0 || dry_run) {
 998       ss << "no change";
 999       cmdctx->reply(r, ss);
1000       return true;
1001     } else {
1002       json_spirit::Object json_object;
1003       for (const auto& osd_weight : new_weights) {
1004         json_spirit::Config::add(json_object,
1005                                  std::to_string(osd_weight.first),
1006                                  std::to_string(osd_weight.second));
1007       }
1008       string s = json_spirit::write(json_object);
1009       std::replace(begin(s), end(s), '\"', '\'');
1010       const string cmd =
1011         "{"
1012         "\"prefix\": \"osd reweightn\", "
1013         "\"weights\": \"" + s + "\""
1014         "}";
1015       auto on_finish = new ReplyOnFinish(cmdctx);
1016       monc->start_mon_command({cmd}, {},
1017                               &on_finish->from_mon, &on_finish->outs, on_finish);
1018       return true;
1019     }
1020   } else if (prefix == "osd df") {
1021     string method;
1022     cmd_getval(g_ceph_context, cmdctx->cmdmap, "output_method", method);
1023     r = cluster_state.with_pgservice([&](const PGMapStatService& pgservice) {
1024         return cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1025             print_osd_utilization(osdmap, &pgservice, ss,
1026                                   f.get(), method == "tree");
1027
1028             cmdctx->odata.append(ss);
1029             return 0;
1030           });
1031       });
1032     cmdctx->reply(r, "");
1033     return true;
1034   } else if (prefix == "osd safe-to-destroy") {
1035     vector<string> ids;
1036     cmd_getval(g_ceph_context, cmdctx->cmdmap, "ids", ids);
1037     set<int> osds;
1038     int r;
1039     cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1040         r = osdmap.parse_osd_id_list(ids, &osds, &ss);
1041       });
1042     if (!r && osds.empty()) {
1043       ss << "must specify one or more OSDs";
1044       r = -EINVAL;
1045     }
1046     if (r < 0) {
1047       cmdctx->reply(r, ss);
1048       return true;
1049     }
1050     set<int> active_osds, missing_stats, stored_pgs;
1051     int affected_pgs = 0;
1052     cluster_state.with_pgmap([&](const PGMap& pg_map) {
1053         if (pg_map.num_pg_unknown > 0) {
1054           ss << pg_map.num_pg_unknown << " pgs have unknown state; cannot draw"
1055              << " any conclusions";
1056           r = -EAGAIN;
1057           return;
1058         }
1059         int num_active_clean = 0;
1060         for (auto& p : pg_map.num_pg_by_state) {
1061           unsigned want = PG_STATE_ACTIVE|PG_STATE_CLEAN;
1062           if ((p.first & want) == want) {
1063             num_active_clean += p.second;
1064           }
1065         }
1066         cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1067             for (auto osd : osds) {
1068               if (!osdmap.exists(osd)) {
1069                 continue;  // clearly safe to destroy
1070               }
1071               auto q = pg_map.num_pg_by_osd.find(osd);
1072               if (q != pg_map.num_pg_by_osd.end()) {
1073                 if (q->second.acting > 0 || q->second.up > 0) {
1074                   active_osds.insert(osd);
1075                   affected_pgs += q->second.acting + q->second.up;
1076                   continue;
1077                 }
1078               }
1079               if (num_active_clean < pg_map.num_pg) {
1080                 // all pgs aren't active+clean; we need to be careful.
1081                 auto p = pg_map.osd_stat.find(osd);
1082                 if (p == pg_map.osd_stat.end()) {
1083                   missing_stats.insert(osd);
1084                 } else if (p->second.num_pgs > 0) {
1085                   stored_pgs.insert(osd);
1086                 }
1087               }
1088             }
1089           });
1090       });
1091     if (!r && !active_osds.empty()) {
1092       ss << "OSD(s) " << active_osds << " have " << affected_pgs
1093          << " pgs currently mapped to them";
1094       r = -EBUSY;
1095     } else if (!missing_stats.empty()) {
1096       ss << "OSD(s) " << missing_stats << " have no reported stats, and not all"
1097          << " PGs are active+clean; we cannot draw any conclusions";
1098       r = -EAGAIN;
1099     } else if (!stored_pgs.empty()) {
1100       ss << "OSD(s) " << stored_pgs << " last reported they still store some PG"
1101          << " data, and not all PGs are active+clean; we cannot be sure they"
1102          << " aren't still needed.";
1103       r = -EBUSY;
1104     }
1105     if (r) {
1106       cmdctx->reply(r, ss);
1107       return true;
1108     }
1109     ss << "OSD(s) " << osds << " are safe to destroy without reducing data"
1110        << " durability.";
1111     cmdctx->reply(0, ss);
1112     return true;
1113   } else if (prefix == "osd ok-to-stop") {
1114     vector<string> ids;
1115     cmd_getval(g_ceph_context, cmdctx->cmdmap, "ids", ids);
1116     set<int> osds;
1117     int r;
1118     cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1119         r = osdmap.parse_osd_id_list(ids, &osds, &ss);
1120       });
1121     if (!r && osds.empty()) {
1122       ss << "must specify one or more OSDs";
1123       r = -EINVAL;
1124     }
1125     if (r < 0) {
1126       cmdctx->reply(r, ss);
1127       return true;
1128     }
1129     map<pg_t,int> pg_delta;  // pgid -> net acting set size change
1130     int dangerous_pgs = 0;
1131     cluster_state.with_pgmap([&](const PGMap& pg_map) {
1132         return cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1133             if (pg_map.num_pg_unknown > 0) {
1134               ss << pg_map.num_pg_unknown << " pgs have unknown state; "
1135                  << "cannot draw any conclusions";
1136               r = -EAGAIN;
1137               return;
1138             }
1139             for (auto osd : osds) {
1140               auto p = pg_map.pg_by_osd.find(osd);
1141               if (p != pg_map.pg_by_osd.end()) {
1142                 for (auto& pgid : p->second) {
1143                   --pg_delta[pgid];
1144                 }
1145               }
1146             }
1147             for (auto& p : pg_delta) {
1148               auto q = pg_map.pg_stat.find(p.first);
1149               if (q == pg_map.pg_stat.end()) {
1150                 ss << "missing information about " << p.first << "; cannot draw"
1151                    << " any conclusions";
1152                 r = -EAGAIN;
1153                 return;
1154               }
1155               if (!(q->second.state & PG_STATE_ACTIVE) ||
1156                   (q->second.state & PG_STATE_DEGRADED)) {
1157                 // we don't currently have a good way to tell *how* degraded
1158                 // a degraded PG is, so we have to assume we cannot remove
1159                 // any more replicas/shards.
1160                 ++dangerous_pgs;
1161                 continue;
1162               }
1163               const pg_pool_t *pi = osdmap.get_pg_pool(p.first.pool());
1164               if (!pi) {
1165                 ++dangerous_pgs; // pool is creating or deleting
1166               } else {
1167                 if (q->second.acting.size() + p.second < pi->min_size) {
1168                   ++dangerous_pgs;
1169                 }
1170               }
1171             }
1172           });
1173       });
1174     if (r) {
1175       cmdctx->reply(r, ss);
1176       return true;
1177     }
1178     if (dangerous_pgs) {
1179       ss << dangerous_pgs << " PGs are already degraded or might become "
1180          << "unavailable";
1181       cmdctx->reply(-EBUSY, ss);
1182       return true;
1183     }
1184     ss << "OSD(s) " << osds << " are ok to stop without reducing"
1185        << " availability, provided there are no other concurrent failures"
1186        << " or interventions. " << pg_delta.size() << " PGs are likely to be"
1187        << " degraded (but remain available) as a result.";
1188     cmdctx->reply(0, ss);
1189     return true;
1190   } else if (prefix == "pg force-recovery" ||
1191                prefix == "pg force-backfill" ||
1192                prefix == "pg cancel-force-recovery" ||
1193                prefix == "pg cancel-force-backfill") {
1194     string forceop = prefix.substr(3, string::npos);
1195     list<pg_t> parsed_pgs;
1196     map<int, list<pg_t> > osdpgs;
1197
1198     // figure out actual op just once
1199     int actual_op = 0;
1200     if (forceop == "force-recovery") {
1201       actual_op = OFR_RECOVERY;
1202     } else if (forceop == "force-backfill") {
1203       actual_op = OFR_BACKFILL;
1204     } else if (forceop == "cancel-force-backfill") {
1205       actual_op = OFR_BACKFILL | OFR_CANCEL;
1206     } else if (forceop == "cancel-force-recovery") {
1207       actual_op = OFR_RECOVERY | OFR_CANCEL;
1208     }
1209
1210     // covnert pg names to pgs, discard any invalid ones while at it
1211     {
1212       // we don't want to keep pgidstr and pgidstr_nodup forever
1213       vector<string> pgidstr;
1214       // get pgids to process and prune duplicates
1215       cmd_getval(g_ceph_context, cmdctx->cmdmap, "pgid", pgidstr);
1216       set<string> pgidstr_nodup(pgidstr.begin(), pgidstr.end());
1217       if (pgidstr.size() != pgidstr_nodup.size()) {
1218         // move elements only when there were duplicates, as this
1219         // reorders them
1220         pgidstr.resize(pgidstr_nodup.size());
1221         auto it = pgidstr_nodup.begin();
1222         for (size_t i = 0 ; i < pgidstr_nodup.size(); i++) {
1223           pgidstr[i] = std::move(*it++);
1224         }
1225       }
1226
1227       cluster_state.with_pgmap([&](const PGMap& pg_map) {
1228         for (auto& pstr : pgidstr) {
1229           pg_t parsed_pg;
1230           if (!parsed_pg.parse(pstr.c_str())) {
1231             ss << "invalid pgid '" << pstr << "'; ";
1232             r = -EINVAL;
1233           } else {
1234             auto workit = pg_map.pg_stat.find(parsed_pg);
1235             if (workit == pg_map.pg_stat.end()) {
1236               ss << "pg " << pstr << " does not exist; ";
1237               r = -ENOENT;
1238             } else {
1239               pg_stat_t workpg = workit->second;
1240
1241               // discard pgs for which user requests are pointless
1242               switch (actual_op)
1243               {
1244                 case OFR_RECOVERY:
1245                   if ((workpg.state & (PG_STATE_DEGRADED | PG_STATE_RECOVERY_WAIT | PG_STATE_RECOVERING)) == 0) {
1246                     // don't return error, user script may be racing with cluster. not fatal.
1247                     ss << "pg " << pstr << " doesn't require recovery; ";
1248                     continue;
1249                   } else  if (workpg.state & PG_STATE_FORCED_RECOVERY) {
1250                     ss << "pg " << pstr << " recovery already forced; ";
1251                     // return error, as it may be a bug in user script
1252                     r = -EINVAL;
1253                     continue;
1254                   }
1255                   break;
1256                 case OFR_BACKFILL:
1257                   if ((workpg.state & (PG_STATE_DEGRADED | PG_STATE_BACKFILL_WAIT | PG_STATE_BACKFILLING)) == 0) {
1258                     ss << "pg " << pstr << " doesn't require backfilling; ";
1259                     continue;
1260                   } else  if (workpg.state & PG_STATE_FORCED_BACKFILL) {
1261                     ss << "pg " << pstr << " backfill already forced; ";
1262                     r = -EINVAL;
1263                     continue;
1264                   }
1265                   break;
1266                 case OFR_BACKFILL | OFR_CANCEL:
1267                   if ((workpg.state & PG_STATE_FORCED_BACKFILL) == 0) {
1268                     ss << "pg " << pstr << " backfill not forced; ";
1269                     continue;
1270                   }
1271                   break;
1272                 case OFR_RECOVERY | OFR_CANCEL:
1273                   if ((workpg.state & PG_STATE_FORCED_RECOVERY) == 0) {
1274                     ss << "pg " << pstr << " recovery not forced; ";
1275                     continue;
1276                   }
1277                   break;
1278                 default:
1279                   assert(0 == "actual_op value is not supported");
1280               }
1281
1282               parsed_pgs.push_back(std::move(parsed_pg));
1283             }
1284           }
1285         }
1286
1287         // group pgs to process by osd
1288         for (auto& pgid : parsed_pgs) {
1289           auto workit = pg_map.pg_stat.find(pgid);
1290           if (workit != pg_map.pg_stat.end()) {
1291             pg_stat_t workpg = workit->second;
1292             set<int32_t> osds(workpg.up.begin(), workpg.up.end());
1293             osds.insert(workpg.acting.begin(), workpg.acting.end());
1294             for (auto i : osds) {
1295               osdpgs[i].push_back(pgid);
1296             }
1297           }
1298         }
1299
1300       });
1301     }
1302
1303     // respond with error only when no pgs are correct
1304     // yes, in case of mixed errors, only the last one will be emitted,
1305     // but the message presented will be fine
1306     if (parsed_pgs.size() != 0) {
1307       // clear error to not confuse users/scripts
1308       r = 0;
1309     }
1310
1311     // optimize the command -> messages conversion, use only one message per distinct OSD
1312     cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1313       for (auto& i : osdpgs) {
1314         if (osdmap.is_up(i.first)) {
1315           vector<pg_t> pgvec(make_move_iterator(i.second.begin()), make_move_iterator(i.second.end()));
1316           auto p = osd_cons.find(i.first);
1317           if (p == osd_cons.end()) {
1318             ss << "osd." << i.first << " is not currently connected";
1319             r = -EAGAIN;
1320             continue;
1321           }
1322           for (auto& con : p->second) {
1323             con->send_message(new MOSDForceRecovery(monc->get_fsid(), pgvec, actual_op));
1324           }
1325           ss << "instructing pg(s) " << i.second << " on osd." << i.first << " to " << forceop << "; ";
1326         }
1327       }
1328     });
1329     ss << std::endl;
1330     cmdctx->reply(r, ss);
1331     return true;
1332   } else {
1333     r = cluster_state.with_pgmap([&](const PGMap& pg_map) {
1334         return cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1335             return process_pg_map_command(prefix, cmdctx->cmdmap, pg_map, osdmap,
1336                                           f.get(), &ss, &cmdctx->odata);
1337           });
1338       });
1339
1340     if (r != -EOPNOTSUPP) {
1341       cmdctx->reply(r, ss);
1342       return true;
1343     }
1344   }
1345
1346   // None of the special native commands,
1347   ActivePyModule *handler = nullptr;
1348   auto py_commands = py_modules.get_py_commands();
1349   for (const auto &pyc : py_commands) {
1350     auto pyc_prefix = cmddesc_get_prefix(pyc.cmdstring);
1351     dout(1) << "pyc_prefix: '" << pyc_prefix << "'" << dendl;
1352     if (pyc_prefix == prefix) {
1353       handler = pyc.handler;
1354       break;
1355     }
1356   }
1357
1358   if (handler == nullptr) {
1359     ss << "No handler found for '" << prefix << "'";
1360     dout(4) << "No handler found for '" << prefix << "'" << dendl;
1361     cmdctx->reply(-EINVAL, ss);
1362     return true;
1363   } else {
1364     // Okay, now we have a handler to call, but we must not call it
1365     // in this thread, because the python handlers can do anything,
1366     // including blocking, and including calling back into mgr.
1367     dout(4) << "passing through " << cmdctx->cmdmap.size() << dendl;
1368     finisher.queue(new FunctionContext([cmdctx, handler](int r_) {
1369       std::stringstream ds;
1370       std::stringstream ss;
1371       int r = handler->handle_command(cmdctx->cmdmap, &ds, &ss);
1372       cmdctx->odata.append(ds);
1373       cmdctx->reply(r, ss);
1374     }));
1375     return true;
1376   }
1377 }
1378
1379 void DaemonServer::_prune_pending_service_map()
1380 {
1381   utime_t cutoff = ceph_clock_now();
1382   cutoff -= g_conf->get_val<double>("mgr_service_beacon_grace");
1383   auto p = pending_service_map.services.begin();
1384   while (p != pending_service_map.services.end()) {
1385     auto q = p->second.daemons.begin();
1386     while (q != p->second.daemons.end()) {
1387       DaemonKey key(p->first, q->first);
1388       if (!daemon_state.exists(key)) {
1389         derr << "missing key " << key << dendl;
1390         ++q;
1391         continue;
1392       }
1393       auto daemon = daemon_state.get(key);
1394       Mutex::Locker l(daemon->lock);
1395       if (daemon->last_service_beacon == utime_t()) {
1396         // we must have just restarted; assume they are alive now.
1397         daemon->last_service_beacon = ceph_clock_now();
1398         ++q;
1399         continue;
1400       }
1401       if (daemon->last_service_beacon < cutoff) {
1402         dout(10) << "pruning stale " << p->first << "." << q->first
1403                  << " last_beacon " << daemon->last_service_beacon << dendl;
1404         q = p->second.daemons.erase(q);
1405         pending_service_map_dirty = pending_service_map.epoch;
1406       } else {
1407         ++q;
1408       }
1409     }
1410     if (p->second.daemons.empty()) {
1411       p = pending_service_map.services.erase(p);
1412       pending_service_map_dirty = pending_service_map.epoch;
1413     } else {
1414       ++p;
1415     }
1416   }
1417 }
1418
1419 void DaemonServer::send_report()
1420 {
1421   if (!pgmap_ready) {
1422     if (ceph_clock_now() - started_at > g_conf->get_val<int64_t>("mgr_stats_period") * 4.0) {
1423       pgmap_ready = true;
1424       reported_osds.clear();
1425       dout(1) << "Giving up on OSDs that haven't reported yet, sending "
1426               << "potentially incomplete PG state to mon" << dendl;
1427     } else {
1428       dout(1) << "Not sending PG status to monitor yet, waiting for OSDs"
1429               << dendl;
1430       return;
1431     }
1432   }
1433
1434   auto m = new MMonMgrReport();
1435   py_modules.get_health_checks(&m->health_checks);
1436
1437   cluster_state.with_pgmap([&](const PGMap& pg_map) {
1438       cluster_state.update_delta_stats();
1439
1440       if (pending_service_map.epoch) {
1441         _prune_pending_service_map();
1442         if (pending_service_map_dirty >= pending_service_map.epoch) {
1443           pending_service_map.modified = ceph_clock_now();
1444           ::encode(pending_service_map, m->service_map_bl, CEPH_FEATURES_ALL);
1445           dout(10) << "sending service_map e" << pending_service_map.epoch
1446                    << dendl;
1447           pending_service_map.epoch++;
1448         }
1449       }
1450
1451       cluster_state.with_osdmap([&](const OSDMap& osdmap) {
1452           // FIXME: no easy way to get mon features here.  this will do for
1453           // now, though, as long as we don't make a backward-incompat change.
1454           pg_map.encode_digest(osdmap, m->get_data(), CEPH_FEATURES_ALL);
1455           dout(10) << pg_map << dendl;
1456
1457           pg_map.get_health_checks(g_ceph_context, osdmap,
1458                                    &m->health_checks);
1459
1460           dout(10) << m->health_checks.checks.size() << " health checks"
1461                    << dendl;
1462           dout(20) << "health checks:\n";
1463           JSONFormatter jf(true);
1464           jf.dump_object("health_checks", m->health_checks);
1465           jf.flush(*_dout);
1466           *_dout << dendl;
1467         });
1468     });
1469
1470   auto osds = daemon_state.get_by_service("osd");
1471   map<osd_metric, unique_ptr<OSDHealthMetricCollector>> accumulated;
1472   for (const auto& osd : osds) {
1473     Mutex::Locker l(osd.second->lock);
1474     for (const auto& metric : osd.second->osd_health_metrics) {
1475       auto acc = accumulated.find(metric.get_type());
1476       if (acc == accumulated.end()) {
1477         auto collector = OSDHealthMetricCollector::create(metric.get_type());
1478         if (!collector) {
1479           derr << __func__ << " " << osd.first << "." << osd.second
1480                << " sent me an unknown health metric: "
1481                << static_cast<uint8_t>(metric.get_type()) << dendl;
1482           continue;
1483         }
1484         tie(acc, std::ignore) = accumulated.emplace(metric.get_type(),
1485                                                     std::move(collector));
1486       }
1487       acc->second->update(osd.first, metric);
1488     }
1489   }
1490   for (const auto& acc : accumulated) {
1491     acc.second->summarize(m->health_checks);
1492   }
1493   // TODO? We currently do not notify the PyModules
1494   // TODO: respect needs_send, so we send the report only if we are asked to do
1495   //       so, or the state is updated.
1496   monc->send_mon_message(m);
1497 }
1498
1499 void DaemonServer::got_service_map()
1500 {
1501   Mutex::Locker l(lock);
1502
1503   cluster_state.with_servicemap([&](const ServiceMap& service_map) {
1504       if (pending_service_map.epoch == 0) {
1505         // we just started up
1506         dout(10) << "got initial map e" << service_map.epoch << dendl;
1507         pending_service_map = service_map;
1508       } else {
1509         // we we already active and therefore must have persisted it,
1510         // which means ours is the same or newer.
1511         dout(10) << "got updated map e" << service_map.epoch << dendl;
1512       }
1513       pending_service_map.epoch = service_map.epoch + 1;
1514     });
1515
1516   // cull missing daemons, populate new ones
1517   for (auto& p : pending_service_map.services) {
1518     std::set<std::string> names;
1519     for (auto& q : p.second.daemons) {
1520       names.insert(q.first);
1521       DaemonKey key(p.first, q.first);
1522       if (!daemon_state.exists(key)) {
1523         auto daemon = std::make_shared<DaemonState>(daemon_state.types);
1524         daemon->key = key;
1525         daemon->metadata = q.second.metadata;
1526         if (q.second.metadata.count("hostname")) {
1527           daemon->hostname = q.second.metadata["hostname"];
1528         }
1529         daemon->service_daemon = true;
1530         daemon_state.insert(daemon);
1531         dout(10) << "added missing " << key << dendl;
1532       }
1533     }
1534     daemon_state.cull(p.first, names);
1535   }
1536 }
1537
1538
1539 const char** DaemonServer::get_tracked_conf_keys() const
1540 {
1541   static const char *KEYS[] = {
1542     "mgr_stats_threshold",
1543     "mgr_stats_period",
1544     nullptr
1545   };
1546
1547   return KEYS;
1548 }
1549
1550 void DaemonServer::handle_conf_change(const struct md_config_t *conf,
1551                                               const std::set <std::string> &changed)
1552 {
1553   dout(4) << "ohai" << dendl;
1554   // We may be called within lock (via MCommand `config set`) or outwith the
1555   // lock (via admin socket `config set`), so handle either case.
1556   const bool initially_locked = lock.is_locked_by_me();
1557   if (!initially_locked) {
1558     lock.Lock();
1559   }
1560
1561   if (changed.count("mgr_stats_threshold") || changed.count("mgr_stats_period")) {
1562     dout(4) << "Updating stats threshold/period on "
1563             << daemon_connections.size() << " clients" << dendl;
1564     // Send a fresh MMgrConfigure to all clients, so that they can follow
1565     // the new policy for transmitting stats
1566     for (auto &c : daemon_connections) {
1567       _send_configure(c);
1568     }
1569   }
1570 }
1571
1572 void DaemonServer::_send_configure(ConnectionRef c)
1573 {
1574   assert(lock.is_locked_by_me());
1575
1576   auto configure = new MMgrConfigure();
1577   configure->stats_period = g_conf->get_val<int64_t>("mgr_stats_period");
1578   configure->stats_threshold = g_conf->get_val<int64_t>("mgr_stats_threshold");
1579   c->send_message(configure);
1580 }
1581