Monitor::Monitor(CephContext* cct_, string nm, MonitorDBStore *s,
		 Messenger *m, Messenger *mgr_m, MonMap *map) :
  Dispatcher(cct_),
  name(nm),
  rank(-1), 
  messenger(m),
  con_self(m ? m->get_loopback_connection() : NULL),
  lock("Monitor::lock"),
  timer(cct_, lock),
  finisher(cct_, "mon_finisher", "fin"),
  cpu_tp(cct, "Monitor::cpu_tp", "cpu_tp", g_conf->mon_cpu_threads),
  has_ever_joined(false),
  logger(NULL), cluster_logger(NULL), cluster_logger_registered(false),
  monmap(map),
  log_client(cct_, messenger, monmap, LogClient::FLAG_MON),
  key_server(cct, &keyring),
  auth_cluster_required(cct,
			cct->_conf->auth_supported.empty() ?
			cct->_conf->auth_cluster_required : cct->_conf->auth_supported),
  auth_service_required(cct,
			cct->_conf->auth_supported.empty() ?
			cct->_conf->auth_service_required : cct->_conf->auth_supported ),
  mgr_messenger(mgr_m),
  mgr_client(cct_, mgr_m),
  pgservice(nullptr),
  store(s),
  
  state(STATE_PROBING),
  
  elector(this),
  required_features(0),
  leader(0),
  quorum_con_features(0),
  // scrub
  scrub_version(0),
  scrub_event(NULL),
  scrub_timeout_event(NULL),

  // sync state
  sync_provider_count(0),
  sync_cookie(0),
  sync_full(false),
  sync_start_version(0),
  sync_timeout_event(NULL),
  sync_last_committed_floor(0),

  timecheck_round(0),
  timecheck_acks(0),
  timecheck_rounds_since_clean(0),
  timecheck_event(NULL),

  paxos_service(PAXOS_NUM),
  admin_hook(NULL),
  routed_request_tid(0),
  op_tracker(cct, true, 1)
{
  clog = log_client.create_channel(CLOG_CHANNEL_CLUSTER);
  audit_clog = log_client.create_channel(CLOG_CHANNEL_AUDIT);

  update_log_clients();

  paxos = new Paxos(this, "paxos");

  paxos_service[PAXOS_MDSMAP] = new MDSMonitor(this, paxos, "mdsmap");
  paxos_service[PAXOS_MONMAP] = new MonmapMonitor(this, paxos, "monmap");
  paxos_service[PAXOS_OSDMAP] = new OSDMonitor(cct, this, paxos, "osdmap");
  paxos_service[PAXOS_PGMAP] = new PGMonitor(this, paxos, "pgmap");
  paxos_service[PAXOS_LOG] = new LogMonitor(this, paxos, "logm");
  paxos_service[PAXOS_AUTH] = new AuthMonitor(this, paxos, "auth");
  paxos_service[PAXOS_MGR] = new MgrMonitor(this, paxos, "mgr");
  paxos_service[PAXOS_MGRSTAT] = new MgrStatMonitor(this, paxos, "mgrstat");
  paxos_service[PAXOS_HEALTH] = new HealthMonitor(this, paxos, "health");

  health_monitor = new OldHealthMonitor(this);
  config_key_service = new ConfigKeyService(this, paxos);

  mon_caps = new MonCap();
  bool r = mon_caps->parse("allow *", NULL);
  assert(r);

  exited_quorum = ceph_clock_now();

  // prepare local commands
  local_mon_commands.resize(ARRAY_SIZE(mon_commands));
  for (unsigned i = 0; i < ARRAY_SIZE(mon_commands); ++i) {
    local_mon_commands[i] = mon_commands[i];
  }
  MonCommand::encode_vector(local_mon_commands, local_mon_commands_bl);

  local_upgrading_mon_commands = local_mon_commands;
  for (unsigned i = 0; i < ARRAY_SIZE(pgmonitor_commands); ++i) {
    local_upgrading_mon_commands.push_back(pgmonitor_commands[i]);
  }
  MonCommand::encode_vector(local_upgrading_mon_commands,
			    local_upgrading_mon_commands_bl);

  // assume our commands until we have an election.  this only means
  // we won't reply with EINVAL before the election; any command that
  // actually matters will wait until we have quorum etc and then
  // retry (and revalidate).
  leader_mon_commands = local_mon_commands;

  // note: OSDMonitor may update this based on the luminous flag.
  pgservice = mgrstatmon()->get_pg_stat_service();
}