ceph/src/mon/Session.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #ifndef CEPH_MON_SESSION_H
  16 #define CEPH_MON_SESSION_H
  17
  18 #include "global/global_context.h"
  19 #include "include/xlist.h"
  20 #include "msg/msg_types.h"
  21 #include "mon/mon_types.h"
  22
  23 #include "auth/AuthServiceHandler.h"
  24 #include "osd/OSDMap.h"
  25
  26 #include "MonCap.h"
  27
  28 struct MonSession;
  29
  30 struct Subscription {
  31   MonSession *session;
  32   string type;
  33   xlist<Subscription*>::item type_item;
  34   version_t next;
  35   bool onetime;
  36   bool incremental_onetime;  // has CEPH_FEATURE_INCSUBOSDMAP
  37
  38   Subscription(MonSession *s, const string& t) : session(s), type(t), type_item(this),
  39                                                  next(0), onetime(false), incremental_onetime(false) {}
  40 };
  41
  42 struct MonSession : public RefCountedObject {
  43   ConnectionRef con;
  44   int con_type = 0;
  45   uint64_t con_features = 0;  // zero if AnonConnection
  46   entity_name_t name;
  47   entity_addrvec_t addrs;
  48   entity_addr_t socket_addr;
  49   utime_t session_timeout;
  50   bool closed = false;
  51   xlist<MonSession*>::item item;
  52   set<uint64_t> routed_request_tids;
  53   MonCap caps;
  54
  55   bool authenticated = false;  ///< true if auth handshake is complete
  56
  57   map<string, Subscription*> sub_map;
  58   epoch_t osd_epoch = 0;       ///< the osdmap epoch sent to the mon client
  59
  60   AuthServiceHandler *auth_handler = nullptr;
  61   EntityName entity_name;
  62
  63   ConnectionRef proxy_con;
  64   uint64_t proxy_tid = 0;
  65
  66   string remote_host;                ///< remote host name
  67   map<string,string> last_config;    ///< most recently shared config
  68   bool any_config = false;
  69
  70   MonSession(Connection *c)
  71     : RefCountedObject(g_ceph_context),
  72       con(c),
  73       item(this) { }
  74
  75   void _ident(const entity_name_t& n, const entity_addrvec_t& av) {
  76     con_type = con->get_peer_type();
  77     name = n;
  78     addrs = av;
  79     socket_addr = con->get_peer_socket_addr();
  80     if (con->get_messenger()) {
  81       // only fill in features if this is a non-anonymous connection
  82       con_features = con->get_features();
  83     }
  84   }
  85
  86   ~MonSession() override {
  87     //generic_dout(0) << "~MonSession " << this << dendl;
  88     // we should have been removed before we get destructed; see MonSessionMap::remove_session()
  89     ceph_assert(!item.is_on_list());
  90     ceph_assert(sub_map.empty());
  91     delete auth_handler;
  92   }
  93
  94   bool is_capable(string service, int mask) {
  95     map<string,string> args;
  96     return caps.is_capable(
  97       g_ceph_context,
  98       entity_name,
  99       service, "", args,
 100       mask & MON_CAP_R, mask & MON_CAP_W, mask & MON_CAP_X,
 101       get_peer_socket_addr());
 102   }
 103
 104   const entity_addr_t& get_peer_socket_addr() {
 105     return socket_addr;
 106   }
 107 };
 108
 109
 110 struct MonSessionMap {
 111   xlist<MonSession*> sessions;
 112   map<string, xlist<Subscription*>* > subs;
 113   multimap<int, MonSession*> by_osd;
 114   FeatureMap feature_map; // type -> features -> count
 115
 116   MonSessionMap() {}
 117   ~MonSessionMap() {
 118     while (!subs.empty()) {
 119       ceph_assert(subs.begin()->second->empty());
 120       delete subs.begin()->second;
 121       subs.erase(subs.begin());
 122     }
 123   }
 124
 125   unsigned get_size() const {
 126     return sessions.size();
 127   }
 128
 129   void remove_session(MonSession *s) {
 130     ceph_assert(!s->closed);
 131     for (map<string,Subscription*>::iterator p = s->sub_map.begin(); p != s->sub_map.end(); ++p) {
 132       p->second->type_item.remove_myself();
 133       delete p->second;
 134     }
 135     s->sub_map.clear();
 136     s->item.remove_myself();
 137     if (s->name.is_osd() &&
 138         s->name.num() >= 0) {
 139       for (multimap<int,MonSession*>::iterator p = by_osd.find(s->name.num());
 140            p->first == s->name.num();
 141            ++p)
 142         if (p->second == s) {
 143           by_osd.erase(p);
 144           break;
 145         }
 146     }
 147     if (s->con_features) {
 148       feature_map.rm(s->con_type, s->con_features);
 149     }
 150     s->closed = true;
 151     s->put();
 152   }
 153
 154   MonSession *new_session(const entity_name_t& n,
 155                           const entity_addrvec_t& av,
 156                           Connection *c) {
 157     MonSession *s = new MonSession(c);
 158     ceph_assert(s);
 159     s->_ident(n, av);
 160     add_session(s);
 161     return s;
 162   }
 163
 164   void add_session(MonSession *s) {
 165     sessions.push_back(&s->item);
 166     s->get();
 167     if (s->name.is_osd() &&
 168         s->name.num() >= 0) {
 169       by_osd.insert(pair<int,MonSession*>(s->name.num(), s));
 170     }
 171     if (s->con_features) {
 172       feature_map.add(s->con_type, s->con_features);
 173     }
 174   }
 175
 176   MonSession *get_random_osd_session(OSDMap *osdmap) {
 177     // ok, this isn't actually random, but close enough.
 178     if (by_osd.empty())
 179       return 0;
 180     int n = by_osd.rbegin()->first + 1;
 181     int r = rand() % n;
 182
 183     multimap<int,MonSession*>::iterator p = by_osd.lower_bound(r);
 184     if (p == by_osd.end())
 185       --p;
 186
 187     if (!osdmap) {
 188       return p->second;
 189     }
 190
 191     MonSession *s = NULL;
 192
 193     multimap<int,MonSession*>::iterator b = p, f = p;
 194     bool backward = true, forward = true;
 195     while (backward || forward) {
 196       if (backward) {
 197         if (osdmap->is_up(b->first) &&
 198             osdmap->get_addrs(b->first) == b->second->con->get_peer_addrs()) {
 199           s = b->second;
 200           break;
 201         }
 202         if (b != by_osd.begin())
 203           --b;
 204         else
 205           backward = false;
 206       }
 207
 208       forward = (f != by_osd.end());
 209       if (forward) {
 210         if (osdmap->is_up(f->first)) {
 211           s = f->second;
 212           break;
 213         }
 214         ++f;
 215       }
 216     }
 217
 218     return s;
 219   }
 220
 221   void add_update_sub(MonSession *s, const string& what, version_t start, bool onetime, bool incremental_onetime) {
 222     Subscription *sub = 0;
 223     if (s->sub_map.count(what)) {
 224       sub = s->sub_map[what];
 225     } else {
 226       sub = new Subscription(s, what);
 227       s->sub_map[what] = sub;
 228
 229       if (!subs.count(what))
 230         subs[what] = new xlist<Subscription*>;
 231       subs[what]->push_back(&sub->type_item);
 232     }
 233     sub->next = start;
 234     sub->onetime = onetime;
 235     sub->incremental_onetime = onetime && incremental_onetime;
 236   }
 237
 238   void remove_sub(Subscription *sub) {
 239     sub->session->sub_map.erase(sub->type);
 240     sub->type_item.remove_myself();
 241     delete sub;
 242   }
 243 };
 244
 245 inline ostream& operator<<(ostream& out, const MonSession& s)
 246 {
 247   out << "MonSession(" << s.name << " " << s.addrs
 248       << " is " << (s.closed ? "closed" : "open")
 249       << " " << s.caps
 250       << ", features 0x" << std::hex << s.con_features << std::dec
 251       <<  " (" << ceph_release_name(ceph_release_from_features(s.con_features))
 252       << "))";
 253   return out;
 254 }
 255
 256 #endif