bool DaemonServer::handle_report(MMgrReport *m)
{
  DaemonKey key;
  if (!m->service_name.empty()) {
    key.first = m->service_name;
  } else {
    key.first = ceph_entity_type_name(m->get_connection()->get_peer_type());
  }
  key.second = m->daemon_name;

  dout(4) << "from " << m->get_connection() << " " << key << dendl;

  if (m->get_connection()->get_peer_type() == entity_name_t::TYPE_CLIENT &&
      m->service_name.empty()) {
    // Clients should not be sending us stats unless they are declaring
    // themselves to be a daemon for some service.
    dout(4) << "rejecting report from non-daemon client " << m->daemon_name
	    << dendl;
    m->get_connection()->mark_down();
    m->put();
    return true;
  }

  // Look up the DaemonState
  DaemonStatePtr daemon;
  if (daemon_state.exists(key)) {
    dout(20) << "updating existing DaemonState for " << key << dendl;
    daemon = daemon_state.get(key);
  } else {
    // we don't know the hostname at this stage, reject MMgrReport here.
    dout(5) << "rejecting report from " << key << ", since we do not have its metadata now."
	    << dendl;

    // issue metadata request in background
    if (!daemon_state.is_updating(key) && 
	(key.first == "osd" || key.first == "mds")) {

      std::ostringstream oss;
      auto c = new MetadataUpdate(daemon_state, key);
      if (key.first == "osd") {
        oss << "{\"prefix\": \"osd metadata\", \"id\": "
            << key.second<< "}";

      } else if (key.first == "mds") {
        c->set_default("addr", stringify(m->get_source_addr()));
        oss << "{\"prefix\": \"mds metadata\", \"who\": \""
            << key.second << "\"}";
 
      } else {
	ceph_abort();
      }

      monc->start_mon_command({oss.str()}, {}, &c->outbl, &c->outs, c);
    }
    
    {
      Mutex::Locker l(lock);
      // kill session
      MgrSessionRef session(static_cast<MgrSession*>(m->get_connection()->get_priv()));
      if (!session) {
	return false;
      }
      m->get_connection()->mark_down();
      session->put();

      dout(10) << "unregistering osd." << session->osd_id
	       << "  session " << session << " con " << m->get_connection() << dendl;
      
      if (osd_cons.find(session->osd_id) != osd_cons.end()) {
	   osd_cons[session->osd_id].erase(m->get_connection());
      } 

      auto iter = daemon_connections.find(m->get_connection());
      if (iter != daemon_connections.end()) {
	daemon_connections.erase(iter);
      }
    }

    return false;
  }

  // Update the DaemonState
  assert(daemon != nullptr);
  {
    Mutex::Locker l(daemon->lock);
    auto &daemon_counters = daemon->perf_counters;
    daemon_counters.update(m);

    if (daemon->service_daemon) {
      utime_t now = ceph_clock_now();
      if (m->daemon_status) {
        daemon->service_status = *m->daemon_status;
        daemon->service_status_stamp = now;
      }
      daemon->last_service_beacon = now;
    } else if (m->daemon_status) {
      derr << "got status from non-daemon " << key << dendl;
    }
    if (m->get_connection()->peer_is_osd()) {
      // only OSD sends health_checks to me now
      daemon->osd_health_metrics = std::move(m->osd_health_metrics);
    }
  }

  // if there are any schema updates, notify the python modules
  if (!m->declare_types.empty() || !m->undeclare_types.empty()) {
    ostringstream oss;
    oss << key.first << '.' << key.second;
    py_modules.notify_all("perf_schema_update", oss.str());
  }

  m->put();
  return true;
}