void OSD::do_command(Connection *con, ceph_tid_t tid, vector<string>& cmd, bufferlist& data)
{
  int r = 0;
  stringstream ss, ds;
  string rs;
  bufferlist odata;

  dout(20) << "do_command tid " << tid << " " << cmd << dendl;

  map<string, cmd_vartype> cmdmap;
  string prefix;
  string format;
  string pgidstr;
  boost::scoped_ptr<Formatter> f;

  if (cmd.empty()) {
    ss << "no command given";
    goto out;
  }

  if (!cmdmap_from_json(cmd, &cmdmap, ss)) {
    r = -EINVAL;
    goto out;
  }

  cmd_getval(cct, cmdmap, "prefix", prefix);

  if (prefix == "get_command_descriptions") {
    int cmdnum = 0;
    JSONFormatter *f = new JSONFormatter();
    f->open_object_section("command_descriptions");
    for (OSDCommand *cp = osd_commands;
	 cp < &osd_commands[ARRAY_SIZE(osd_commands)]; cp++) {

      ostringstream secname;
      secname << "cmd" << setfill('0') << std::setw(3) << cmdnum;
      dump_cmddesc_to_json(f, secname.str(), cp->cmdstring, cp->helpstring,
			   cp->module, cp->perm, cp->availability, 0);
      cmdnum++;
    }
    f->close_section();	// command_descriptions

    f->flush(ds);
    delete f;
    goto out;
  }

  cmd_getval(cct, cmdmap, "format", format);
  f.reset(Formatter::create(format));

  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();
    }
    goto out;
  }
  else if (prefix == "injectargs") {
    vector<string> argsvec;
    cmd_getval(cct, cmdmap, "injected_args", argsvec);

    if (argsvec.empty()) {
      r = -EINVAL;
      ss << "ignoring empty injectargs";
      goto out;
    }
    string args = argsvec.front();
    for (vector<string>::iterator a = ++argsvec.begin(); a != argsvec.end(); ++a)
      args += " " + *a;
    osd_lock.Unlock();
    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);
    if (msg.empty()) {
      r = -EINVAL;
      ss << "ignoring empty log message";
      goto out;
    }
    string message = msg.front();
    for (vector<string>::iterator a = ++msg.begin(); a != msg.end(); ++a)
      message += " " + *a;
    string lvl;
    cmd_getval(cct, cmdmap, "level", lvl);
    clog_type level = string_to_clog_type(lvl);
    if (level < 0) {
      r = -EINVAL;
      ss << "unknown level '" << lvl << "'";
      goto out;
    }
    clog->do_log(level, message);
  }

  // either 'pg <pgid> <command>' or
  // 'tell <pgid>' (which comes in without any of that prefix)?

  else if (prefix == "pg" ||
	    prefix == "query" ||
	    prefix == "mark_unfound_lost" ||
	    prefix == "list_missing"
	   ) {
    pg_t pgid;

    if (!cmd_getval(cct, cmdmap, "pgid", pgidstr)) {
      ss << "no pgid specified";
      r = -EINVAL;
    } else if (!pgid.parse(pgidstr.c_str())) {
      ss << "couldn't parse pgid '" << pgidstr << "'";
      r = -EINVAL;
    } else {
      spg_t pcand;
      PG *pg = nullptr;
      if (osdmap->get_primary_shard(pgid, &pcand) &&
	  (pg = _lookup_lock_pg(pcand))) {
	if (pg->is_primary()) {
	  // simulate pg <pgid> cmd= for pg->do-command
	  if (prefix != "pg")
	    cmd_putval(cct, cmdmap, "cmd", prefix);
	  r = pg->do_command(cmdmap, ss, data, odata, con, tid);
	  if (r == -EAGAIN) {
	    pg->unlock();
	    // don't reply, pg will do so async
	    return;
	  }
	} else {
	  ss << "not primary for pgid " << pgid;

	  // send them the latest diff to ensure they realize the mapping
	  // has changed.
	  service.send_incremental_map(osdmap->get_epoch() - 1, con, osdmap);

	  // do not reply; they will get newer maps and realize they
	  // need to resend.
	  pg->unlock();
	  return;
	}
	pg->unlock();
      } else {
	ss << "i don't have pgid " << pgid;
	r = -ENOENT;
      }
    }
  }

  else if (prefix == "bench") {
    int64_t count;
    int64_t bsize;
    int64_t osize, onum;
    // default count 1G, size 4MB
    cmd_getval(cct, cmdmap, "count", count, (int64_t)1 << 30);
    cmd_getval(cct, cmdmap, "size", bsize, (int64_t)4 << 20);
    cmd_getval(cct, cmdmap, "object_size", osize, (int64_t)0);
    cmd_getval(cct, cmdmap, "object_num", onum, (int64_t)0);

    ceph::shared_ptr<ObjectStore::Sequencer> osr (std::make_shared<
                                        ObjectStore::Sequencer>("bench"));

    uint32_t duration = cct->_conf->osd_bench_duration;

    if (bsize > (int64_t) cct->_conf->osd_bench_max_block_size) {
      // let us limit the block size because the next checks rely on it
      // having a sane value.  If we allow any block size to be set things
      // can still go sideways.
      ss << "block 'size' values are capped at "
         << prettybyte_t(cct->_conf->osd_bench_max_block_size) << ". If you wish to use"
         << " a higher value, please adjust 'osd_bench_max_block_size'";
      r = -EINVAL;
      goto out;
    } else if (bsize < (int64_t) (1 << 20)) {
      // entering the realm of small block sizes.
      // limit the count to a sane value, assuming a configurable amount of
      // IOPS and duration, so that the OSD doesn't get hung up on this,
      // preventing timeouts from going off
      int64_t max_count =
        bsize * duration * cct->_conf->osd_bench_small_size_max_iops;
      if (count > max_count) {
        ss << "'count' values greater than " << max_count
           << " for a block size of " << prettybyte_t(bsize) << ", assuming "
           << cct->_conf->osd_bench_small_size_max_iops << " IOPS,"
           << " for " << duration << " seconds,"
           << " can cause ill effects on osd. "
           << " Please adjust 'osd_bench_small_size_max_iops' with a higher"
           << " value if you wish to use a higher 'count'.";
        r = -EINVAL;
        goto out;
      }
    } else {
      // 1MB block sizes are big enough so that we get more stuff done.
      // However, to avoid the osd from getting hung on this and having
      // timers being triggered, we are going to limit the count assuming
      // a configurable throughput and duration.
      // NOTE: max_count is the total amount of bytes that we believe we
      //       will be able to write during 'duration' for the given
      //       throughput.  The block size hardly impacts this unless it's
      //       way too big.  Given we already check how big the block size
      //       is, it's safe to assume everything will check out.
      int64_t max_count =
        cct->_conf->osd_bench_large_size_max_throughput * duration;
      if (count > max_count) {
        ss << "'count' values greater than " << max_count
           << " for a block size of " << prettybyte_t(bsize) << ", assuming "
           << prettybyte_t(cct->_conf->osd_bench_large_size_max_throughput) << "/s,"
           << " for " << duration << " seconds,"
           << " can cause ill effects on osd. "
           << " Please adjust 'osd_bench_large_size_max_throughput'"
           << " with a higher value if you wish to use a higher 'count'.";
        r = -EINVAL;
        goto out;
      }
    }

    if (osize && bsize > osize)
      bsize = osize;

    dout(1) << " bench count " << count
            << " bsize " << prettybyte_t(bsize) << dendl;

    ObjectStore::Transaction cleanupt;

    if (osize && onum) {
      bufferlist bl;
      bufferptr bp(osize);
      bp.zero();
      bl.push_back(std::move(bp));
      bl.rebuild_page_aligned();
      for (int i=0; i<onum; ++i) {
	char nm[30];
	snprintf(nm, sizeof(nm), "disk_bw_test_%d", i);
	object_t oid(nm);
	hobject_t soid(sobject_t(oid, 0));
	ObjectStore::Transaction t;
	t.write(coll_t(), ghobject_t(soid), 0, osize, bl);
	store->queue_transaction(osr.get(), std::move(t), NULL);
	cleanupt.remove(coll_t(), ghobject_t(soid));
      }
    }

    bufferlist bl;
    bufferptr bp(bsize);
    bp.zero();
    bl.push_back(std::move(bp));
    bl.rebuild_page_aligned();

    {
      C_SaferCond waiter;
      if (!osr->flush_commit(&waiter)) {
	waiter.wait();
      }
    }

    utime_t start = ceph_clock_now();
    for (int64_t pos = 0; pos < count; pos += bsize) {
      char nm[30];
      unsigned offset = 0;
      if (onum && osize) {
	snprintf(nm, sizeof(nm), "disk_bw_test_%d", (int)(rand() % onum));
	offset = rand() % (osize / bsize) * bsize;
      } else {
	snprintf(nm, sizeof(nm), "disk_bw_test_%lld", (long long)pos);
      }
      object_t oid(nm);
      hobject_t soid(sobject_t(oid, 0));
      ObjectStore::Transaction t;
      t.write(coll_t::meta(), ghobject_t(soid), offset, bsize, bl);
      store->queue_transaction(osr.get(), std::move(t), NULL);
      if (!onum || !osize)
	cleanupt.remove(coll_t::meta(), ghobject_t(soid));
    }

    {
      C_SaferCond waiter;
      if (!osr->flush_commit(&waiter)) {
	waiter.wait();
      }
    }
    utime_t end = ceph_clock_now();

    // clean up
    store->queue_transaction(osr.get(), std::move(cleanupt), NULL);
    {
      C_SaferCond waiter;
      if (!osr->flush_commit(&waiter)) {
	waiter.wait();
      }
    }

    uint64_t rate = (double)count / (end - start);
    if (f) {
      f->open_object_section("osd_bench_results");
      f->dump_int("bytes_written", count);
      f->dump_int("blocksize", bsize);
      f->dump_unsigned("bytes_per_sec", rate);
      f->close_section();
      f->flush(ss);
    } else {
      ss << "bench: wrote " << prettybyte_t(count)
	 << " in blocks of " << prettybyte_t(bsize) << " in "
	 << (end-start) << " sec at " << prettybyte_t(rate) << "/sec";
    }
  }

  else if (prefix == "flush_pg_stats") {
    if (osdmap->require_osd_release >= CEPH_RELEASE_LUMINOUS) {
      mgrc.send_pgstats();
      ds << service.get_osd_stat_seq() << "\n";
    } else {
      flush_pg_stats();
    }
  }

  else if (prefix == "heap") {
    r = ceph::osd_cmds::heap(*cct, cmdmap, *f, ds);
  }

  else if (prefix == "debug dump_missing") {
    string file_name;
    cmd_getval(cct, cmdmap, "filename", file_name);
    std::ofstream fout(file_name.c_str());
    if (!fout.is_open()) {
	ss << "failed to open file '" << file_name << "'";
	r = -EINVAL;
	goto out;
    }

    fout << "*** osd " << whoami << ": dump_missing ***" << std::endl;
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t, PG*>::const_iterator pg_map_e = pg_map.begin();
	 pg_map_e != pg_map.end(); ++pg_map_e) {
      PG *pg = pg_map_e->second;
      pg->lock();

      fout << *pg << std::endl;
      std::map<hobject_t, pg_missing_item>::const_iterator mend =
	pg->pg_log.get_missing().get_items().end();
      std::map<hobject_t, pg_missing_item>::const_iterator mi =
	pg->pg_log.get_missing().get_items().begin();
      for (; mi != mend; ++mi) {
	fout << mi->first << " -> " << mi->second << std::endl;
	if (!pg->missing_loc.needs_recovery(mi->first))
	  continue;
	if (pg->missing_loc.is_unfound(mi->first))
	  fout << " unfound ";
	const set<pg_shard_t> &mls(pg->missing_loc.get_locations(mi->first));
	if (mls.empty())
	  continue;
	fout << "missing_loc: " << mls << std::endl;
      }
      pg->unlock();
      fout << std::endl;
    }

    fout.close();
  }
  else if (prefix == "debug kick_recovery_wq") {
    int64_t delay;
    cmd_getval(cct, cmdmap, "delay", delay);
    ostringstream oss;
    oss << delay;
    r = cct->_conf->set_val("osd_recovery_delay_start", oss.str().c_str());
    if (r != 0) {
      ss << "kick_recovery_wq: error setting "
	 << "osd_recovery_delay_start to '" << delay << "': error "
	 << r;
      goto out;
    }
    cct->_conf->apply_changes(NULL);
    ss << "kicking recovery queue. set osd_recovery_delay_start "
       << "to " << cct->_conf->osd_recovery_delay_start;
  }

  else if (prefix == "cpu_profiler") {
    string arg;
    cmd_getval(cct, cmdmap, "arg", arg);
    vector<string> argvec;
    get_str_vec(arg, argvec);
    cpu_profiler_handle_command(argvec, ds);
  }

  else if (prefix == "dump_pg_recovery_stats") {
    stringstream s;
    if (f) {
      pg_recovery_stats.dump_formatted(f.get());
      f->flush(ds);
    } else {
      pg_recovery_stats.dump(s);
      ds << "dump pg recovery stats: " << s.str();
    }
  }

  else if (prefix == "reset_pg_recovery_stats") {
    ss << "reset pg recovery stats";
    pg_recovery_stats.reset();
  }

  else if (prefix == "perf histogram dump") {
    std::string logger;
    std::string counter;
    cmd_getval(cct, cmdmap, "logger", logger);
    cmd_getval(cct, cmdmap, "counter", counter);
    if (f) {
      cct->get_perfcounters_collection()->dump_formatted_histograms(
          f.get(), false, logger, counter);
      f->flush(ds);
    }
  }

  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;
  }

 out:
  rs = ss.str();
  odata.append(ds);
  dout(0) << "do_command r=" << r << " " << rs << dendl;
  clog->info() << rs;
  if (con) {
    MCommandReply *reply = new MCommandReply(r, rs);
    reply->set_tid(tid);
    reply->set_data(odata);
    con->send_message(reply);
  }
}