void Monitor::handle_command(MonOpRequestRef op)
{
  assert(op->is_type_command());
  MMonCommand *m = static_cast<MMonCommand*>(op->get_req());
  if (m->fsid != monmap->fsid) {
    dout(0) << "handle_command on fsid " << m->fsid << " != " << monmap->fsid << dendl;
    reply_command(op, -EPERM, "wrong fsid", 0);
    return;
  }

  MonSession *session = static_cast<MonSession *>(
    m->get_connection()->get_priv());
  if (!session) {
    dout(5) << __func__ << " dropping stray message " << *m << dendl;
    return;
  }
  BOOST_SCOPE_EXIT_ALL(=) {
    session->put();
  };

  if (m->cmd.empty()) {
    string rs = "No command supplied";
    reply_command(op, -EINVAL, rs, 0);
    return;
  }

  string prefix;
  vector<string> fullcmd;
  map<string, cmd_vartype> cmdmap;
  stringstream ss, ds;
  bufferlist rdata;
  string rs;
  int r = -EINVAL;
  rs = "unrecognized command";

  if (!cmdmap_from_json(m->cmd, &cmdmap, ss)) {
    // ss has reason for failure
    r = -EINVAL;
    rs = ss.str();
    if (!m->get_source().is_mon())  // don't reply to mon->mon commands
      reply_command(op, r, rs, 0);
    return;
  }

  // check return value. If no prefix parameter provided,
  // return value will be false, then return error info.
  if (!cmd_getval(g_ceph_context, cmdmap, "prefix", prefix)) {
    reply_command(op, -EINVAL, "command prefix not found", 0);
    return;
  }

  // check prefix is empty
  if (prefix.empty()) {
    reply_command(op, -EINVAL, "command prefix must not be empty", 0);
    return;
  }

  if (prefix == "get_command_descriptions") {
    bufferlist rdata;
    Formatter *f = Formatter::create("json");
    // hide mgr commands until luminous upgrade is complete
    bool hide_mgr_flag =
      osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS;

    std::vector<MonCommand> commands;

    // only include mgr commands once all mons are upgrade (and we've dropped
    // the hard-coded PGMonitor commands)
    if (quorum_mon_features.contains_all(ceph::features::mon::FEATURE_LUMINOUS)) {
      commands = static_cast<MgrMonitor*>(
        paxos_service[PAXOS_MGR])->get_command_descs();
    }

    for (auto& c : leader_mon_commands) {
      commands.push_back(c);
    }

    format_command_descriptions(commands, f, &rdata, hide_mgr_flag);
    delete f;
    reply_command(op, 0, "", rdata, 0);
    return;
  }

  string module;
  string err;

  dout(0) << "handle_command " << *m << dendl;

  string format;
  cmd_getval(g_ceph_context, cmdmap, "format", format, string("plain"));
  boost::scoped_ptr<Formatter> f(Formatter::create(format));

  get_str_vec(prefix, fullcmd);

  // make sure fullcmd is not empty.
  // invalid prefix will cause empty vector fullcmd.
  // such as, prefix=";,,;"
  if (fullcmd.empty()) {
    reply_command(op, -EINVAL, "command requires a prefix to be valid", 0);
    return;
  }

  module = fullcmd[0];

  // validate command is in leader map

  const MonCommand *leader_cmd;
  const auto& mgr_cmds = mgrmon()->get_command_descs();
  const MonCommand *mgr_cmd = nullptr;
  if (!mgr_cmds.empty()) {
    mgr_cmd = _get_moncommand(prefix, mgr_cmds);
  }
  leader_cmd = _get_moncommand(prefix, leader_mon_commands);
  if (!leader_cmd) {
    leader_cmd = mgr_cmd;
    if (!leader_cmd) {
      reply_command(op, -EINVAL, "command not known", 0);
      return;
    }
  }
  // validate command is in our map & matches, or forward if it is allowed
  const MonCommand *mon_cmd = _get_moncommand(
    prefix,
    get_local_commands(quorum_mon_features));
  if (!mon_cmd) {
    mon_cmd = mgr_cmd;
  }
  if (!is_leader()) {
    if (!mon_cmd) {
      if (leader_cmd->is_noforward()) {
	reply_command(op, -EINVAL,
		      "command not locally supported and not allowed to forward",
		      0);
	return;
      }
      dout(10) << "Command not locally supported, forwarding request "
	       << m << dendl;
      forward_request_leader(op);
      return;
    } else if (!mon_cmd->is_compat(leader_cmd)) {
      if (mon_cmd->is_noforward()) {
	reply_command(op, -EINVAL,
		      "command not compatible with leader and not allowed to forward",
		      0);
	return;
      }
      dout(10) << "Command not compatible with leader, forwarding request "
	       << m << dendl;
      forward_request_leader(op);
      return;
    }
  }

  if (mon_cmd->is_obsolete() ||
      (cct->_conf->mon_debug_deprecated_as_obsolete
       && mon_cmd->is_deprecated())) {
    reply_command(op, -ENOTSUP,
                  "command is obsolete; please check usage and/or man page",
                  0);
    return;
  }

  if (session->proxy_con && mon_cmd->is_noforward()) {
    dout(10) << "Got forward for noforward command " << m << dendl;
    reply_command(op, -EINVAL, "forward for noforward command", rdata, 0);
    return;
  }

  /* what we perceive as being the service the command falls under */
  string service(mon_cmd->module);

  dout(25) << __func__ << " prefix='" << prefix
           << "' module='" << module
           << "' service='" << service << "'" << dendl;

  bool cmd_is_rw =
    (mon_cmd->requires_perm('w') || mon_cmd->requires_perm('x'));

  // validate user's permissions for requested command
  map<string,string> param_str_map;
  _generate_command_map(cmdmap, param_str_map);
  if (!_allowed_command(session, service, prefix, cmdmap,
                        param_str_map, mon_cmd)) {
    dout(1) << __func__ << " access denied" << dendl;
    (cmd_is_rw ? audit_clog->info() : audit_clog->debug())
      << "from='" << session->inst << "' "
      << "entity='" << session->entity_name << "' "
      << "cmd=" << m->cmd << ":  access denied";
    reply_command(op, -EACCES, "access denied", 0);
    return;
  }

  (cmd_is_rw ? audit_clog->info() : audit_clog->debug())
    << "from='" << session->inst << "' "
    << "entity='" << session->entity_name << "' "
    << "cmd=" << m->cmd << ": dispatch";

  if (mon_cmd->is_mgr() &&
      osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
    const auto& hdr = m->get_header();
    uint64_t size = hdr.front_len + hdr.middle_len + hdr.data_len;
    uint64_t max = g_conf->get_val<uint64_t>("mon_client_bytes")
                 * g_conf->get_val<double>("mon_mgr_proxy_client_bytes_ratio");
    if (mgr_proxy_bytes + size > max) {
      dout(10) << __func__ << " current mgr proxy bytes " << mgr_proxy_bytes
	       << " + " << size << " > max " << max << dendl;
      reply_command(op, -EAGAIN, "hit limit on proxied mgr commands", rdata, 0);
      return;
    }
    mgr_proxy_bytes += size;
    dout(10) << __func__ << " proxying mgr command (+" << size
	     << " -> " << mgr_proxy_bytes << ")" << dendl;
    C_MgrProxyCommand *fin = new C_MgrProxyCommand(this, op, size);
    mgr_client.start_command(m->cmd,
			     m->get_data(),
			     &fin->outbl,
			     &fin->outs,
			     new C_OnFinisher(fin, &finisher));
    return;
  }

  if ((module == "mds" || module == "fs")  &&
      prefix != "fs authorize") {
    mdsmon()->dispatch(op);
    return;
  }
  if ((module == "osd" || prefix == "pg map") &&
      prefix != "osd last-stat-seq") {
    osdmon()->dispatch(op);
    return;
  }

  if (module == "pg") {
    pgmon()->dispatch(op);
    return;
  }
  if (module == "mon" &&
      /* Let the Monitor class handle the following commands:
       *  'mon compact'
       *  'mon scrub'
       *  'mon sync force'
       */
      prefix != "mon compact" &&
      prefix != "mon scrub" &&
      prefix != "mon sync force" &&
      prefix != "mon metadata" &&
      prefix != "mon versions" &&
      prefix != "mon count-metadata") {
    monmon()->dispatch(op);
    return;
  }
  if (module == "auth" || prefix == "fs authorize") {
    authmon()->dispatch(op);
    return;
  }
  if (module == "log") {
    logmon()->dispatch(op);
    return;
  }

  if (module == "config-key") {
    config_key_service->dispatch(op);
    return;
  }

  if (module == "mgr") {
    mgrmon()->dispatch(op);
    return;
  }

  if (prefix == "fsid") {
    if (f) {
      f->open_object_section("fsid");
      f->dump_stream("fsid") << monmap->fsid;
      f->close_section();
      f->flush(rdata);
    } else {
      ds << monmap->fsid;
      rdata.append(ds);
    }
    reply_command(op, 0, "", rdata, 0);
    return;
  }

  if (prefix == "scrub" || prefix == "mon scrub") {
    wait_for_paxos_write();
    if (is_leader()) {
      int r = scrub_start();
      reply_command(op, r, "", rdata, 0);
    } else if (is_peon()) {
      forward_request_leader(op);
    } else {
      reply_command(op, -EAGAIN, "no quorum", rdata, 0);
    }
    return;
  }

  if (prefix == "compact" || prefix == "mon compact") {
    dout(1) << "triggering manual compaction" << dendl;
    utime_t start = ceph_clock_now();
    store->compact();
    utime_t end = ceph_clock_now();
    end -= start;
    dout(1) << "finished manual compaction in " << end << " seconds" << dendl;
    ostringstream oss;
    oss << "compacted " << g_conf->get_val<std::string>("mon_keyvaluedb") << " in " << end << " seconds";
    rs = oss.str();
    r = 0;
  }
  else if (prefix == "injectargs") {
    vector<string> injected_args;
    cmd_getval(g_ceph_context, cmdmap, "injected_args", injected_args);
    if (!injected_args.empty()) {
      dout(0) << "parsing injected options '" << injected_args << "'" << dendl;
      ostringstream oss;
      r = g_conf->injectargs(str_join(injected_args, " "), &oss);
      ss << "injectargs:"  << oss.str();
      rs = ss.str();
      goto out;
    } else {
      rs = "must supply options to be parsed in a single string";
      r = -EINVAL;
    }
  } else if (prefix == "time-sync-status") {
    if (!f)
      f.reset(Formatter::create("json-pretty"));
    f->open_object_section("time_sync");
    if (!timecheck_skews.empty()) {
      f->open_object_section("time_skew_status");
      for (auto& i : timecheck_skews) {
	entity_inst_t inst = i.first;
	double skew = i.second;
	double latency = timecheck_latencies[inst];
	string name = monmap->get_name(inst.addr);
	ostringstream tcss;
	health_status_t tcstatus = timecheck_status(tcss, skew, latency);
	f->open_object_section(name.c_str());
	f->dump_float("skew", skew);
	f->dump_float("latency", latency);
	f->dump_stream("health") << tcstatus;
	if (tcstatus != HEALTH_OK) {
	  f->dump_stream("details") << tcss.str();
	}
	f->close_section();
      }
      f->close_section();
    }
    f->open_object_section("timechecks");
    f->dump_unsigned("epoch", get_epoch());
    f->dump_int("round", timecheck_round);
    f->dump_stream("round_status") << ((timecheck_round%2) ?
				       "on-going" : "finished");
    f->close_section();
    f->close_section();
    f->flush(rdata);
    r = 0;
    rs = "";
  } else if (prefix == "config set") {
    std::string key;
    cmd_getval(cct, cmdmap, "key", key);
    std::string val;
    cmd_getval(cct, cmdmap, "value", val);
    r = g_conf->set_val(key, val, true, &ss);
    if (r == 0) {
      g_conf->apply_changes(nullptr);
    }
    rs = ss.str();
    goto out;
  } else if (prefix == "status" ||
	     prefix == "health" ||
	     prefix == "df") {
    string detail;
    cmd_getval(g_ceph_context, cmdmap, "detail", detail);

    if (prefix == "status") {
      // get_cluster_status handles f == NULL
      get_cluster_status(ds, f.get());

      if (f) {
        f->flush(ds);
        ds << '\n';
      }
      rdata.append(ds);
    } else if (prefix == "health") {
      if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
	string plain;
	get_health_status(detail == "detail", f.get(), f ? nullptr : &plain);
	if (f) {
	  f->flush(rdata);
	} else {
	  rdata.append(plain);
	}
      } else {
	list<string> health_str;
	get_health(health_str, detail == "detail" ? &rdata : NULL, f.get());
	if (f) {
	  f->flush(ds);
	  ds << '\n';
	} else {
	  assert(!health_str.empty());
	  ds << health_str.front();
	  health_str.pop_front();
	  if (!health_str.empty()) {
	    ds << ' ';
	    ds << joinify(health_str.begin(), health_str.end(), string("; "));
	  }
	}
	bufferlist comb;
	comb.append(ds);
	if (detail == "detail")
	  comb.append(rdata);
	rdata = comb;
      }
    } else if (prefix == "df") {
      bool verbose = (detail == "detail");
      if (f)
        f->open_object_section("stats");

      pgservice->dump_fs_stats(&ds, f.get(), verbose);
      if (!f)
        ds << '\n';
      pgservice->dump_pool_stats(osdmon()->osdmap, &ds, f.get(), verbose);

      if (f) {
        f->close_section();
        f->flush(ds);
        ds << '\n';
      }
    } else {
      assert(0 == "We should never get here!");
      return;
    }
    rdata.append(ds);
    rs = "";
    r = 0;
  } else if (prefix == "report") {

    // this must be formatted, in its current form
    if (!f)
      f.reset(Formatter::create("json-pretty"));
    f->open_object_section("report");
    f->dump_stream("cluster_fingerprint") << fingerprint;
    f->dump_string("version", ceph_version_to_str());
    f->dump_string("commit", git_version_to_str());
    f->dump_stream("timestamp") << ceph_clock_now();

    vector<string> tagsvec;
    cmd_getval(g_ceph_context, cmdmap, "tags", tagsvec);
    string tagstr = str_join(tagsvec, " ");
    if (!tagstr.empty())
      tagstr = tagstr.substr(0, tagstr.find_last_of(' '));
    f->dump_string("tag", tagstr);

    if (osdmon()->osdmap.require_osd_release >= CEPH_RELEASE_LUMINOUS) {
      get_health_status(true, f.get(), nullptr);
    } else {
      list<string> health_str;
      get_health(health_str, nullptr, f.get());
    }

    monmon()->dump_info(f.get());
    osdmon()->dump_info(f.get());
    mdsmon()->dump_info(f.get());
    authmon()->dump_info(f.get());
    pgservice->dump_info(f.get());

    paxos->dump_info(f.get());

    f->close_section();
    f->flush(rdata);

    ostringstream ss2;
    ss2 << "report " << rdata.crc32c(CEPH_MON_PORT);
    rs = ss2.str();
    r = 0;
  } else if (prefix == "osd last-stat-seq") {
    int64_t osd;
    cmd_getval(g_ceph_context, cmdmap, "id", osd);
    uint64_t seq = mgrstatmon()->get_last_osd_stat_seq(osd);
    if (f) {
      f->dump_unsigned("seq", seq);
      f->flush(ds);
    } else {
      ds << seq;
      rdata.append(ds);
    }
    rs = "";
    r = 0;
  } else if (prefix == "node ls") {
    string node_type("all");
    cmd_getval(g_ceph_context, cmdmap, "type", node_type);
    if (!f)
      f.reset(Formatter::create("json-pretty"));
    if (node_type == "all") {
      f->open_object_section("nodes");
      print_nodes(f.get(), ds);
      osdmon()->print_nodes(f.get());
      mdsmon()->print_nodes(f.get());
      f->close_section();
    } else if (node_type == "mon") {
      print_nodes(f.get(), ds);
    } else if (node_type == "osd") {
      osdmon()->print_nodes(f.get());
    } else if (node_type == "mds") {
      mdsmon()->print_nodes(f.get());
    }
    f->flush(ds);
    rdata.append(ds);
    rs = "";
    r = 0;
  } else if (prefix == "features") {
    if (!is_leader() && !is_peon()) {
      dout(10) << " waiting for quorum" << dendl;
      waitfor_quorum.push_back(new C_RetryMessage(this, op));
      return;
    }
    if (!is_leader()) {
      forward_request_leader(op);
      return;
    }
    if (!f)
      f.reset(Formatter::create("json-pretty"));
    FeatureMap fm;
    get_combined_feature_map(&fm);
    f->dump_object("features", fm);
    f->flush(rdata);
    rs = "";
    r = 0;
  } else if (prefix == "mon metadata") {
    if (!f)
      f.reset(Formatter::create("json-pretty"));

    string name;
    bool all = !cmd_getval(g_ceph_context, cmdmap, "id", name);
    if (!all) {
      // Dump a single mon's metadata
      int mon = monmap->get_rank(name);
      if (mon < 0) {
        rs = "requested mon not found";
        r = -ENOENT;
        goto out;
      }
      f->open_object_section("mon_metadata");
      r = get_mon_metadata(mon, f.get(), ds);
      f->close_section();
    } else {
      // Dump all mons' metadata
      r = 0;
      f->open_array_section("mon_metadata");
      for (unsigned int rank = 0; rank < monmap->size(); ++rank) {
        std::ostringstream get_err;
        f->open_object_section("mon");
        f->dump_string("name", monmap->get_name(rank));
        r = get_mon_metadata(rank, f.get(), get_err);
        f->close_section();
        if (r == -ENOENT || r == -EINVAL) {
          dout(1) << get_err.str() << dendl;
          // Drop error, list what metadata we do have
          r = 0;
        } else if (r != 0) {
          derr << "Unexpected error from get_mon_metadata: "
               << cpp_strerror(r) << dendl;
          ds << get_err.str();
          break;
        }
      }
      f->close_section();
    }

    f->flush(ds);
    rdata.append(ds);
    rs = "";
  } else if (prefix == "mon versions") {
    if (!f)
      f.reset(Formatter::create("json-pretty"));
    count_metadata("ceph_version", f.get());
    f->flush(ds);
    rdata.append(ds);
    rs = "";
    r = 0;
  } else if (prefix == "mon count-metadata") {
    if (!f)
      f.reset(Formatter::create("json-pretty"));
    string field;
    cmd_getval(g_ceph_context, cmdmap, "property", field);
    count_metadata(field, f.get());
    f->flush(ds);
    rdata.append(ds);
    rs = "";
    r = 0;
  } else if (prefix == "quorum_status") {
    // make sure our map is readable and up to date
    if (!is_leader() && !is_peon()) {
      dout(10) << " waiting for quorum" << dendl;
      waitfor_quorum.push_back(new C_RetryMessage(this, op));
      return;
    }
    _quorum_status(f.get(), ds);
    rdata.append(ds);
    rs = "";
    r = 0;
  } else if (prefix == "mon_status") {
    get_mon_status(f.get(), ds);
    if (f)
      f->flush(ds);
    rdata.append(ds);
    rs = "";
    r = 0;
  } else if (prefix == "sync force" ||
             prefix == "mon sync force") {
    string validate1, validate2;
    cmd_getval(g_ceph_context, cmdmap, "validate1", validate1);
    cmd_getval(g_ceph_context, cmdmap, "validate2", validate2);
    if (validate1 != "--yes-i-really-mean-it" ||
	validate2 != "--i-know-what-i-am-doing") {
      r = -EINVAL;
      rs = "are you SURE? this will mean the monitor store will be "
	   "erased.  pass '--yes-i-really-mean-it "
	   "--i-know-what-i-am-doing' if you really do.";
      goto out;
    }
    sync_force(f.get(), ds);
    rs = ds.str();
    r = 0;
  } else if (prefix == "heap") {
    if (!ceph_using_tcmalloc())
      rs = "tcmalloc not enabled, can't use heap profiler commands\n";
    else {
      string heapcmd;
      cmd_getval(g_ceph_context, cmdmap, "heapcmd", heapcmd);
      // XXX 1-element vector, change at callee or make vector here?
      vector<string> heapcmd_vec;
      get_str_vec(heapcmd, heapcmd_vec);
      ceph_heap_profiler_handle_command(heapcmd_vec, ds);
      rdata.append(ds);
      rs = "";
      r = 0;
    }
  } else if (prefix == "quorum") {
    string quorumcmd;
    cmd_getval(g_ceph_context, cmdmap, "quorumcmd", quorumcmd);
    if (quorumcmd == "exit") {
      start_election();
      elector.stop_participating();
      rs = "stopped responding to quorum, initiated new election";
      r = 0;
    } else if (quorumcmd == "enter") {
      elector.start_participating();
      start_election();
      rs = "started responding to quorum, initiated new election";
      r = 0;
    } else {
      rs = "needs a valid 'quorum' command";
      r = -EINVAL;
    }
  } else if (prefix == "version") {
    if (f) {
      f->open_object_section("version");
      f->dump_string("version", pretty_version_to_str());
      f->close_section();
      f->flush(ds);
    } else {
      ds << pretty_version_to_str();
    }
    rdata.append(ds);
    rs = "";
    r = 0;
  } else if (prefix == "versions") {
    if (!f)
      f.reset(Formatter::create("json-pretty"));
    map<string,int> overall;
    f->open_object_section("version");
    map<string,int> mon, mgr, osd, mds;

    count_metadata("ceph_version", &mon);
    f->open_object_section("mon");
    for (auto& p : mon) {
      f->dump_int(p.first.c_str(), p.second);
      overall[p.first] += p.second;
    }
    f->close_section();

    mgrmon()->count_metadata("ceph_version", &mgr);
    f->open_object_section("mgr");
    for (auto& p : mgr) {
      f->dump_int(p.first.c_str(), p.second);
      overall[p.first] += p.second;
    }
    f->close_section();

    osdmon()->count_metadata("ceph_version", &osd);
    f->open_object_section("osd");
    for (auto& p : osd) {
      f->dump_int(p.first.c_str(), p.second);
      overall[p.first] += p.second;
    }
    f->close_section();

    mdsmon()->count_metadata("ceph_version", &mds);
    f->open_object_section("mds");
    for (auto& p : mds) {
      f->dump_int(p.first.c_str(), p.second);
      overall[p.first] += p.second;
    }
    f->close_section();

    for (auto& p : mgrstatmon()->get_service_map().services) {
      f->open_object_section(p.first.c_str());
      map<string,int> m;
      p.second.count_metadata("ceph_version", &m);
      for (auto& q : m) {
	f->dump_int(q.first.c_str(), q.second);
	overall[q.first] += q.second;
      }
      f->close_section();
    }

    f->open_object_section("overall");
    for (auto& p : overall) {
      f->dump_int(p.first.c_str(), p.second);
    }
    f->close_section();
    f->close_section();
    f->flush(rdata);
    rs = "";
    r = 0;
  }

 out:
  if (!m->get_source().is_mon())  // don't reply to mon->mon commands
    reply_command(op, r, rs, rdata, 0);
}