bool OSD::asok_command(string admin_command, cmdmap_t& cmdmap, string format,
		       ostream& ss)
{
  Formatter *f = Formatter::create(format, "json-pretty", "json-pretty");
  if (admin_command == "status") {
    f->open_object_section("status");
    f->dump_stream("cluster_fsid") << superblock.cluster_fsid;
    f->dump_stream("osd_fsid") << superblock.osd_fsid;
    f->dump_unsigned("whoami", superblock.whoami);
    f->dump_string("state", get_state_name(get_state()));
    f->dump_unsigned("oldest_map", superblock.oldest_map);
    f->dump_unsigned("newest_map", superblock.newest_map);
    {
      RWLock::RLocker l(pg_map_lock);
      f->dump_unsigned("num_pgs", pg_map.size());
    }
    f->close_section();
  } else if (admin_command == "flush_journal") {
    store->flush_journal();
  } else if (admin_command == "dump_ops_in_flight" ||
             admin_command == "ops" ||
             admin_command == "dump_blocked_ops" ||
             admin_command == "dump_historic_ops" ||
             admin_command == "dump_historic_ops_by_duration" ||
             admin_command == "dump_historic_slow_ops") {

    const string error_str = "op_tracker tracking is not enabled now, so no ops are tracked currently, \
even those get stuck. Please enable \"osd_enable_op_tracker\", and the tracker \
will start to track new ops received afterwards.";

    set<string> filters;
    vector<string> filter_str;
    if (cmd_getval(cct, cmdmap, "filterstr", filter_str)) {
        copy(filter_str.begin(), filter_str.end(),
           inserter(filters, filters.end()));
    }

    if (admin_command == "dump_ops_in_flight" ||
        admin_command == "ops") {
      if (!op_tracker.dump_ops_in_flight(f, false, filters)) {
        ss << error_str;
      }
    }
    if (admin_command == "dump_blocked_ops") {
      if (!op_tracker.dump_ops_in_flight(f, true, filters)) {
        ss << error_str;
      }
    }
    if (admin_command == "dump_historic_ops") {
      if (!op_tracker.dump_historic_ops(f, false, filters)) {
        ss << error_str;
      }
    }
    if (admin_command == "dump_historic_ops_by_duration") {
      if (!op_tracker.dump_historic_ops(f, true, filters)) {
        ss << error_str;
      }
    }
    if (admin_command == "dump_historic_slow_ops") {
      if (!op_tracker.dump_historic_slow_ops(f, filters)) {
        ss << error_str;
      }
    }
  } else if (admin_command == "dump_op_pq_state") {
    f->open_object_section("pq");
    op_shardedwq.dump(f);
    f->close_section();
  } else if (admin_command == "dump_blacklist") {
    list<pair<entity_addr_t,utime_t> > bl;
    OSDMapRef curmap = service.get_osdmap();

    f->open_array_section("blacklist");
    curmap->get_blacklist(&bl);
    for (list<pair<entity_addr_t,utime_t> >::iterator it = bl.begin();
	it != bl.end(); ++it) {
      f->open_object_section("entry");
      f->open_object_section("entity_addr_t");
      it->first.dump(f);
      f->close_section(); //entity_addr_t
      it->second.localtime(f->dump_stream("expire_time"));
      f->close_section(); //entry
    }
    f->close_section(); //blacklist
  } else if (admin_command == "dump_watchers") {
    list<obj_watch_item_t> watchers;
    // scan pg's
    {
      Mutex::Locker l(osd_lock);
      RWLock::RLocker l2(pg_map_lock);
      for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();
          it != pg_map.end();
          ++it) {

        list<obj_watch_item_t> pg_watchers;
        PG *pg = it->second;
        pg->lock();
        pg->get_watchers(pg_watchers);
        pg->unlock();
        watchers.splice(watchers.end(), pg_watchers);
      }
    }

    f->open_array_section("watchers");
    for (list<obj_watch_item_t>::iterator it = watchers.begin();
	it != watchers.end(); ++it) {

      f->open_object_section("watch");

      f->dump_string("namespace", it->obj.nspace);
      f->dump_string("object", it->obj.oid.name);

      f->open_object_section("entity_name");
      it->wi.name.dump(f);
      f->close_section(); //entity_name_t

      f->dump_unsigned("cookie", it->wi.cookie);
      f->dump_unsigned("timeout", it->wi.timeout_seconds);

      f->open_object_section("entity_addr_t");
      it->wi.addr.dump(f);
      f->close_section(); //entity_addr_t

      f->close_section(); //watch
    }

    f->close_section(); //watchers
  } else if (admin_command == "dump_reservations") {
    f->open_object_section("reservations");
    f->open_object_section("local_reservations");
    service.local_reserver.dump(f);
    f->close_section();
    f->open_object_section("remote_reservations");
    service.remote_reserver.dump(f);
    f->close_section();
    f->close_section();
  } else if (admin_command == "get_latest_osdmap") {
    get_latest_osdmap();
  } else if (admin_command == "heap") {
    auto result = ceph::osd_cmds::heap(*cct, cmdmap, *f, ss);

    // Note: Failed heap profile commands won't necessarily trigger an error:
    f->open_object_section("result");
    f->dump_string("error", cpp_strerror(result));
    f->dump_bool("success", result >= 0);
    f->close_section();
  } else if (admin_command == "set_heap_property") {
    string property;
    int64_t value = 0;
    string error;
    bool success = false;
    if (!cmd_getval(cct, cmdmap, "property", property)) {
      error = "unable to get property";
      success = false;
    } else if (!cmd_getval(cct, cmdmap, "value", value)) {
      error = "unable to get value";
      success = false;
    } else if (value < 0) {
      error = "negative value not allowed";
      success = false;
    } else if (!ceph_heap_set_numeric_property(property.c_str(), (size_t)value)) {
      error = "invalid property";
      success = false;
    } else {
      success = true;
    }
    f->open_object_section("result");
    f->dump_string("error", error);
    f->dump_bool("success", success);
    f->close_section();
  } else if (admin_command == "get_heap_property") {
    string property;
    size_t value = 0;
    string error;
    bool success = false;
    if (!cmd_getval(cct, cmdmap, "property", property)) {
      error = "unable to get property";
      success = false;
    } else if (!ceph_heap_get_numeric_property(property.c_str(), &value)) {
      error = "invalid property";
      success = false;
    } else {
      success = true;
    }
    f->open_object_section("result");
    f->dump_string("error", error);
    f->dump_bool("success", success);
    f->dump_int("value", value);
    f->close_section();
  } else if (admin_command == "dump_objectstore_kv_stats") {
    store->get_db_statistics(f);
  } else if (admin_command == "dump_scrubs") {
    service.dumps_scrub(f);
  } else if (admin_command == "calc_objectstore_db_histogram") {
    store->generate_db_histogram(f);
  } else if (admin_command == "flush_store_cache") {
    store->flush_cache();
  } else if (admin_command == "dump_pgstate_history") {
    f->open_object_section("pgstate_history");
    RWLock::RLocker l2(pg_map_lock);
    for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();
        it != pg_map.end();
        ++it) {

      PG *pg = it->second;
      f->dump_stream("pg") << pg->get_pgid();
      pg->lock();
      pg->pgstate_history.dump(f);
      pg->unlock();
    }
    f->close_section();
  } else if (admin_command == "compact") {
    dout(1) << "triggering manual compaction" << dendl;
    auto start = ceph::coarse_mono_clock::now();
    store->compact();
    auto end = ceph::coarse_mono_clock::now();
    auto time_span = chrono::duration_cast<chrono::duration<double>>(end - start);
    dout(1) << "finished manual compaction in " 
            << time_span.count()
            << " seconds" << dendl;
    f->open_object_section("compact_result");
    f->dump_float("elapsed_time", time_span.count());
    f->close_section();
  } else {
    assert(0 == "broken asok registration");
  }
  f->flush(ss);
  delete f;
  return true;
}