int Monitor::preinit()
{
  lock.Lock();

  dout(1) << "preinit fsid " << monmap->fsid << dendl;

  int r = sanitize_options();
  if (r < 0) {
    derr << "option sanitization failed!" << dendl;
    lock.Unlock();
    return r;
  }

  assert(!logger);
  {
    PerfCountersBuilder pcb(g_ceph_context, "mon", l_mon_first, l_mon_last);
    pcb.add_u64(l_mon_num_sessions, "num_sessions", "Open sessions", "sess",
        PerfCountersBuilder::PRIO_USEFUL);
    pcb.add_u64_counter(l_mon_session_add, "session_add", "Created sessions",
        "sadd", PerfCountersBuilder::PRIO_INTERESTING);
    pcb.add_u64_counter(l_mon_session_rm, "session_rm", "Removed sessions",
        "srm", PerfCountersBuilder::PRIO_INTERESTING);
    pcb.add_u64_counter(l_mon_session_trim, "session_trim", "Trimmed sessions",
        "strm", PerfCountersBuilder::PRIO_USEFUL);
    pcb.add_u64_counter(l_mon_num_elections, "num_elections", "Elections participated in",
        "ecnt", PerfCountersBuilder::PRIO_USEFUL);
    pcb.add_u64_counter(l_mon_election_call, "election_call", "Elections started",
        "estt", PerfCountersBuilder::PRIO_INTERESTING);
    pcb.add_u64_counter(l_mon_election_win, "election_win", "Elections won",
        "ewon", PerfCountersBuilder::PRIO_INTERESTING);
    pcb.add_u64_counter(l_mon_election_lose, "election_lose", "Elections lost",
        "elst", PerfCountersBuilder::PRIO_INTERESTING);
    logger = pcb.create_perf_counters();
    cct->get_perfcounters_collection()->add(logger);
  }

  assert(!cluster_logger);
  {
    PerfCountersBuilder pcb(g_ceph_context, "cluster", l_cluster_first, l_cluster_last);
    pcb.add_u64(l_cluster_num_mon, "num_mon", "Monitors");
    pcb.add_u64(l_cluster_num_mon_quorum, "num_mon_quorum", "Monitors in quorum");
    pcb.add_u64(l_cluster_num_osd, "num_osd", "OSDs");
    pcb.add_u64(l_cluster_num_osd_up, "num_osd_up", "OSDs that are up");
    pcb.add_u64(l_cluster_num_osd_in, "num_osd_in", "OSD in state \"in\" (they are in cluster)");
    pcb.add_u64(l_cluster_osd_epoch, "osd_epoch", "Current epoch of OSD map");
    pcb.add_u64(l_cluster_osd_bytes, "osd_bytes", "Total capacity of cluster");
    pcb.add_u64(l_cluster_osd_bytes_used, "osd_bytes_used", "Used space");
    pcb.add_u64(l_cluster_osd_bytes_avail, "osd_bytes_avail", "Available space");
    pcb.add_u64(l_cluster_num_pool, "num_pool", "Pools");
    pcb.add_u64(l_cluster_num_pg, "num_pg", "Placement groups");
    pcb.add_u64(l_cluster_num_pg_active_clean, "num_pg_active_clean", "Placement groups in active+clean state");
    pcb.add_u64(l_cluster_num_pg_active, "num_pg_active", "Placement groups in active state");
    pcb.add_u64(l_cluster_num_pg_peering, "num_pg_peering", "Placement groups in peering state");
    pcb.add_u64(l_cluster_num_object, "num_object", "Objects");
    pcb.add_u64(l_cluster_num_object_degraded, "num_object_degraded", "Degraded (missing replicas) objects");
    pcb.add_u64(l_cluster_num_object_misplaced, "num_object_misplaced", "Misplaced (wrong location in the cluster) objects");
    pcb.add_u64(l_cluster_num_object_unfound, "num_object_unfound", "Unfound objects");
    pcb.add_u64(l_cluster_num_bytes, "num_bytes", "Size of all objects");
    pcb.add_u64(l_cluster_num_mds_up, "num_mds_up", "MDSs that are up");
    pcb.add_u64(l_cluster_num_mds_in, "num_mds_in", "MDS in state \"in\" (they are in cluster)");
    pcb.add_u64(l_cluster_num_mds_failed, "num_mds_failed", "Failed MDS");
    pcb.add_u64(l_cluster_mds_epoch, "mds_epoch", "Current epoch of MDS map");
    cluster_logger = pcb.create_perf_counters();
  }

  paxos->init_logger();

  // verify cluster_uuid
  {
    int r = check_fsid();
    if (r == -ENOENT)
      r = write_fsid();
    if (r < 0) {
      lock.Unlock();
      return r;
    }
  }

  // open compatset
  read_features();

  // have we ever joined a quorum?
  has_ever_joined = (store->get(MONITOR_NAME, "joined") != 0);
  dout(10) << "has_ever_joined = " << (int)has_ever_joined << dendl;

  if (!has_ever_joined) {
    // impose initial quorum restrictions?
    list<string> initial_members;
    get_str_list(g_conf->mon_initial_members, initial_members);

    if (!initial_members.empty()) {
      dout(1) << " initial_members " << initial_members << ", filtering seed monmap" << dendl;

      monmap->set_initial_members(g_ceph_context, initial_members, name, messenger->get_myaddr(),
				  &extra_probe_peers);

      dout(10) << " monmap is " << *monmap << dendl;
      dout(10) << " extra probe peers " << extra_probe_peers << dendl;
    }
  } else if (!monmap->contains(name)) {
    derr << "not in monmap and have been in a quorum before; "
         << "must have been removed" << dendl;
    if (g_conf->mon_force_quorum_join) {
      dout(0) << "we should have died but "
              << "'mon_force_quorum_join' is set -- allowing boot" << dendl;
    } else {
      derr << "commit suicide!" << dendl;
      lock.Unlock();
      return -ENOENT;
    }
  }

  {
    // We have a potentially inconsistent store state in hands. Get rid of it
    // and start fresh.
    bool clear_store = false;
    if (store->exists("mon_sync", "in_sync")) {
      dout(1) << __func__ << " clean up potentially inconsistent store state"
	      << dendl;
      clear_store = true;
    }

    if (store->get("mon_sync", "force_sync") > 0) {
      dout(1) << __func__ << " force sync by clearing store state" << dendl;
      clear_store = true;
    }

    if (clear_store) {
      set<string> sync_prefixes = get_sync_targets_names();
      store->clear(sync_prefixes);
    }
  }

  sync_last_committed_floor = store->get("mon_sync", "last_committed_floor");
  dout(10) << "sync_last_committed_floor " << sync_last_committed_floor << dendl;

  init_paxos();
  health_monitor->init();

  if (is_keyring_required()) {
    // we need to bootstrap authentication keys so we can form an
    // initial quorum.
    if (authmon()->get_last_committed() == 0) {
      dout(10) << "loading initial keyring to bootstrap authentication for mkfs" << dendl;
      bufferlist bl;
      int err = store->get("mkfs", "keyring", bl);
      if (err == 0 && bl.length() > 0) {
        // Attempt to decode and extract keyring only if it is found.
        KeyRing keyring;
        bufferlist::iterator p = bl.begin();
        ::decode(keyring, p);
        extract_save_mon_key(keyring);
      }
    }

    string keyring_loc = g_conf->mon_data + "/keyring";

    r = keyring.load(cct, keyring_loc);
    if (r < 0) {
      EntityName mon_name;
      mon_name.set_type(CEPH_ENTITY_TYPE_MON);
      EntityAuth mon_key;
      if (key_server.get_auth(mon_name, mon_key)) {
	dout(1) << "copying mon. key from old db to external keyring" << dendl;
	keyring.add(mon_name, mon_key);
	bufferlist bl;
	keyring.encode_plaintext(bl);
	write_default_keyring(bl);
      } else {
	derr << "unable to load initial keyring " << g_conf->keyring << dendl;
	lock.Unlock();
	return r;
      }
    }
  }

  admin_hook = new AdminHook(this);
  AdminSocket* admin_socket = cct->get_admin_socket();

  // unlock while registering to avoid mon_lock -> admin socket lock dependency.
  lock.Unlock();
  r = admin_socket->register_command("mon_status", "mon_status", admin_hook,
				     "show current monitor status");
  assert(r == 0);
  r = admin_socket->register_command("quorum_status", "quorum_status",
				     admin_hook, "show current quorum status");
  assert(r == 0);
  r = admin_socket->register_command("sync_force",
				     "sync_force name=validate,"
				     "type=CephChoices,"
			             "strings=--yes-i-really-mean-it",
				     admin_hook,
				     "force sync of and clear monitor store");
  assert(r == 0);
  r = admin_socket->register_command("add_bootstrap_peer_hint",
				     "add_bootstrap_peer_hint name=addr,"
				     "type=CephIPAddr",
				     admin_hook,
				     "add peer address as potential bootstrap"
				     " peer for cluster bringup");
  assert(r == 0);
  r = admin_socket->register_command("quorum enter", "quorum enter",
                                     admin_hook,
                                     "force monitor back into quorum");
  assert(r == 0);
  r = admin_socket->register_command("quorum exit", "quorum exit",
                                     admin_hook,
                                     "force monitor out of the quorum");
  assert(r == 0);
  r = admin_socket->register_command("ops",
                                     "ops",
                                     admin_hook,
                                     "show the ops currently in flight");
  assert(r == 0);
  r = admin_socket->register_command("sessions",
                                     "sessions",
                                     admin_hook,
                                     "list existing sessions");
  assert(r == 0);

  lock.Lock();

  // add ourselves as a conf observer
  g_conf->add_observer(this);

  lock.Unlock();
  return 0;
}