path: root/src/mon/Session.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software 
 * Foundation.  See file COPYING.
 * 
 */

#ifndef CEPH_MON_SESSION_H
#define CEPH_MON_SESSION_H

#include "global/global_context.h"
#include "include/xlist.h"
#include "msg/msg_types.h"
#include "mon/mon_types.h"

#include "auth/AuthServiceHandler.h"
#include "osd/OSDMap.h"

#include "MonCap.h"

struct MonSession;

struct Subscription {
  MonSession *session;
  string type;
  xlist<Subscription*>::item type_item;
  version_t next;
  bool onetime;
  bool incremental_onetime;  // has CEPH_FEATURE_INCSUBOSDMAP
  
  Subscription(MonSession *s, const string& t) : session(s), type(t), type_item(this),
						 next(0), onetime(false), incremental_onetime(false) {}
};

struct MonSession : public RefCountedObject {
  ConnectionRef con;
  int con_type = 0;
  uint64_t con_features = 0;  // zero if AnonConnection
  entity_name_t name;
  entity_addrvec_t addrs;
  entity_addr_t socket_addr;
  utime_t session_timeout;
  bool closed = false;
  xlist<MonSession*>::item item;
  set<uint64_t> routed_request_tids;
  MonCap caps;

  bool authenticated = false;  ///< true if auth handshake is complete

  map<string, Subscription*> sub_map;
  epoch_t osd_epoch = 0;       ///< the osdmap epoch sent to the mon client

  AuthServiceHandler *auth_handler = nullptr;
  EntityName entity_name;
  uint64_t global_id = 0;
  global_id_status_t global_id_status = global_id_status_t::NONE;

  ConnectionRef proxy_con;
  uint64_t proxy_tid = 0;

  string remote_host;                ///< remote host name
  map<string,string> last_config;    ///< most recently shared config
  bool any_config = false;

  MonSession(Connection *c)
    : RefCountedObject(g_ceph_context),
      con(c),
      item(this) { }

  void _ident(const entity_name_t& n, const entity_addrvec_t& av) {
    con_type = con->get_peer_type();
    name = n;
    addrs = av;
    socket_addr = con->get_peer_socket_addr();
    if (con->get_messenger()) {
      // only fill in features if this is a non-anonymous connection
      con_features = con->get_features();
    }
  }

  ~MonSession() override {
    //generic_dout(0) << "~MonSession " << this << dendl;
    // we should have been removed before we get destructed; see MonSessionMap::remove_session()
    ceph_assert(!item.is_on_list());
    ceph_assert(sub_map.empty());
    delete auth_handler;
  }

  bool is_capable(string service, int mask) {
    map<string,string> args;
    return caps.is_capable(
      g_ceph_context,
      entity_name,
      service, "", args,
      mask & MON_CAP_R, mask & MON_CAP_W, mask & MON_CAP_X,
      get_peer_socket_addr());
  }

  const entity_addr_t& get_peer_socket_addr() {
    return socket_addr;
  }

  void dump(Formatter *f) const {
    f->dump_stream("name") << name;
    f->dump_stream("entity_name") << entity_name;
    f->dump_object("addrs", addrs);
    f->dump_object("socket_addr", socket_addr);
    f->dump_string("con_type", ceph_entity_type_name(con_type));
    f->dump_unsigned("con_features", con_features);
    f->dump_stream("con_features_hex") << std::hex << con_features << std::dec;
    f->dump_string("con_features_release",
		   ceph_release_name(ceph_release_from_features(con_features)));
    f->dump_bool("open", !closed);
    f->dump_object("caps", caps);
    f->dump_bool("authenticated", authenticated);
    f->dump_unsigned("global_id", global_id);
    f->dump_stream("global_id_status") << global_id_status;
    f->dump_unsigned("osd_epoch", osd_epoch);
    f->dump_string("remote_host", remote_host);
  }
};


struct MonSessionMap {
  xlist<MonSession*> sessions;
  map<string, xlist<Subscription*>* > subs;
  multimap<int, MonSession*> by_osd;
  FeatureMap feature_map; // type -> features -> count

  MonSessionMap() {}
  ~MonSessionMap() {
    while (!subs.empty()) {
      ceph_assert(subs.begin()->second->empty());
      delete subs.begin()->second;
      subs.erase(subs.begin());
    }
  }

  unsigned get_size() const {
    return sessions.size();
  }

  void remove_session(MonSession *s) {
    ceph_assert(!s->closed);
    for (map<string,Subscription*>::iterator p = s->sub_map.begin(); p != s->sub_map.end(); ++p) {
      p->second->type_item.remove_myself();
      delete p->second;
    }
    s->sub_map.clear();
    s->item.remove_myself();
    if (s->name.is_osd() &&
	s->name.num() >= 0) {
      for (multimap<int,MonSession*>::iterator p = by_osd.find(s->name.num());
	   p->first == s->name.num();
	   ++p)
	if (p->second == s) {
	  by_osd.erase(p);
	  break;
	}
    }
    if (s->con_features) {
      feature_map.rm(s->con_type, s->con_features);
    }
    s->closed = true;
    s->put();
  }

  MonSession *new_session(const entity_name_t& n,
			  const entity_addrvec_t& av,
			  Connection *c) {
    MonSession *s = new MonSession(c);
    ceph_assert(s);
    s->_ident(n, av);
    add_session(s);
    return s;
  }

  void add_session(MonSession *s) {
    s->session_timeout = ceph_clock_now();
    s->session_timeout += g_conf()->mon_session_timeout;

    sessions.push_back(&s->item);
    s->get();
    if (s->name.is_osd() &&
	s->name.num() >= 0) {
      by_osd.insert(pair<int,MonSession*>(s->name.num(), s));
    }
    if (s->con_features) {
      feature_map.add(s->con_type, s->con_features);
    }
  }

  MonSession *get_random_osd_session(OSDMap *osdmap) {
    // ok, this isn't actually random, but close enough.
    if (by_osd.empty())
      return 0;
    int n = by_osd.rbegin()->first + 1;
    int r = rand() % n;

    multimap<int,MonSession*>::iterator p = by_osd.lower_bound(r);
    if (p == by_osd.end())
      --p;

    if (!osdmap) {
      return p->second;
    }

    MonSession *s = NULL;

    multimap<int,MonSession*>::iterator b = p, f = p;
    bool backward = true, forward = true;
    while (backward || forward) {
      if (backward) {
        if (osdmap->is_up(b->first) &&
	    osdmap->get_addrs(b->first) == b->second->con->get_peer_addrs()) {
          s = b->second;
          break;
        }
        if (b != by_osd.begin())
          --b;
        else
          backward = false;
      }

      forward = (f != by_osd.end());
      if (forward) {
        if (osdmap->is_up(f->first)) {
          s = f->second;
          break;
        }
        ++f;
      }
    }

    return s;
  }

  void add_update_sub(MonSession *s, const string& what, version_t start, bool onetime, bool incremental_onetime) {
    Subscription *sub = 0;
    if (s->sub_map.count(what)) {
      sub = s->sub_map[what];
    } else {
      sub = new Subscription(s, what);
      s->sub_map[what] = sub;

      if (!subs.count(what))
	subs[what] = new xlist<Subscription*>;
      subs[what]->push_back(&sub->type_item);
    }
    sub->next = start;
    sub->onetime = onetime;
    sub->incremental_onetime = onetime && incremental_onetime;
  }

  void remove_sub(Subscription *sub) {
    sub->session->sub_map.erase(sub->type);
    sub->type_item.remove_myself();
    delete sub;
  }
};

inline ostream& operator<<(ostream& out, const MonSession& s)
{
  out << "MonSession(" << s.name << " " << s.addrs
      << " is " << (s.closed ? "closed" : "open")
      << " " << s.caps
      << ", features 0x" << std::hex << s.con_features << std::dec
      <<  " (" << ceph_release_name(ceph_release_from_features(s.con_features))
      << "))";
  return out;
}

#endif