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 <string>
  19 #include <string_view>
  20
  21 #include "include/utime.h"
  22 #include "include/xlist.h"
  23
  24 #include "global/global_context.h"
  25 #include "msg/msg_types.h"
  26 #include "mon/mon_types.h"
  27
  28 #include "auth/AuthServiceHandler.h"
  29 #include "osd/OSDMap.h"
  30
  31 #include "MonCap.h"
  32
  33 struct MonSession;
  34
  35 struct Subscription {
  36   MonSession *session;
  37   std::string type;
  38   xlist<Subscription*>::item type_item;
  39   version_t next;
  40   bool onetime;
  41   bool incremental_onetime;  // has CEPH_FEATURE_INCSUBOSDMAP
  42
  43   Subscription(MonSession *s, const std::string& t) : session(s), type(t), type_item(this),
  44                                                  next(0), onetime(false), incremental_onetime(false) {}
  45 };
  46
  47 struct MonSession : public RefCountedObject {
  48   ConnectionRef con;
  49   int con_type = 0;
  50   uint64_t con_features = 0;  // zero if AnonConnection
  51   entity_name_t name;
  52   entity_addrvec_t addrs;
  53   entity_addr_t socket_addr;
  54   utime_t session_timeout;
  55   bool closed = false;
  56   xlist<MonSession*>::item item;
  57   std::set<uint64_t> routed_request_tids;
  58   MonCap caps;
  59   bool validated_stretch_connection = false;
  60
  61   bool authenticated = false;  ///< true if auth handshake is complete
  62
  63   std::map<std::string, Subscription*> sub_map;
  64   epoch_t osd_epoch = 0;       ///< the osdmap epoch sent to the mon client
  65
  66   AuthServiceHandler *auth_handler = nullptr;
  67   EntityName entity_name;
  68   uint64_t global_id = 0;
  69   global_id_status_t global_id_status = global_id_status_t::NONE;
  70
  71   ConnectionRef proxy_con;
  72   uint64_t proxy_tid = 0;
  73
  74   std::string remote_host;                ///< remote host name
  75   std::map<std::string,std::string,std::less<>> last_config;    ///< most recently shared config
  76   bool any_config = false;
  77
  78   MonSession(Connection *c)
  79     : RefCountedObject(g_ceph_context),
  80       con(c),
  81       item(this) { }
  82
  83   void _ident(const entity_name_t& n, const entity_addrvec_t& av) {
  84     con_type = con->get_peer_type();
  85     name = n;
  86     addrs = av;
  87     socket_addr = con->get_peer_socket_addr();
  88     if (con->get_messenger()) {
  89       // only fill in features if this is a non-anonymous connection
  90       con_features = con->get_features();
  91     }
  92   }
  93
  94   ~MonSession() override {
  95     //generic_dout(0) << "~MonSession " << this << dendl;
  96     // we should have been removed before we get destructed; see MonSessionMap::remove_session()
  97     ceph_assert(!item.is_on_list());
  98     ceph_assert(sub_map.empty());
  99     delete auth_handler;
 100   }
 101
 102   bool is_capable(std::string service, int mask) {
 103     std::map<std::string,std::string> args;
 104     return caps.is_capable(
 105       g_ceph_context,
 106       entity_name,
 107       service, "", args,
 108       mask & MON_CAP_R, mask & MON_CAP_W, mask & MON_CAP_X,
 109       get_peer_socket_addr());
 110   }
 111
 112   std::vector<string> get_allowed_fs_names() const {
 113     return caps.allowed_fs_names();
 114   }
 115
 116   bool fs_name_capable(string_view fsname, __u8 mask) {
 117     return caps.fs_name_capable(entity_name, fsname, mask);
 118   }
 119
 120   const entity_addr_t& get_peer_socket_addr() {
 121     return socket_addr;
 122   }
 123
 124   void dump(ceph::Formatter *f) const {
 125     f->dump_stream("name") << name;
 126     f->dump_stream("entity_name") << entity_name;
 127     f->dump_object("addrs", addrs);
 128     f->dump_object("socket_addr", socket_addr);
 129     f->dump_string("con_type", ceph_entity_type_name(con_type));
 130     f->dump_unsigned("con_features", con_features);
 131     f->dump_stream("con_features_hex") << std::hex << con_features << std::dec;
 132     f->dump_string("con_features_release",
 133                    ceph_release_name(ceph_release_from_features(con_features)));
 134     f->dump_bool("open", !closed);
 135     f->dump_object("caps", caps);
 136     f->dump_bool("authenticated", authenticated);
 137     f->dump_unsigned("global_id", global_id);
 138     f->dump_stream("global_id_status") << global_id_status;
 139     f->dump_unsigned("osd_epoch", osd_epoch);
 140     f->dump_string("remote_host", remote_host);
 141   }
 142 };
 143
 144
 145 struct MonSessionMap {
 146   xlist<MonSession*> sessions;
 147   std::map<std::string, xlist<Subscription*>* > subs;
 148   std::multimap<int, MonSession*> by_osd;
 149   FeatureMap feature_map; // type -> features -> count
 150
 151   MonSessionMap() {}
 152   ~MonSessionMap() {
 153     while (!subs.empty()) {
 154       ceph_assert(subs.begin()->second->empty());
 155       delete subs.begin()->second;
 156       subs.erase(subs.begin());
 157     }
 158   }
 159
 160   unsigned get_size() const {
 161     return sessions.size();
 162   }
 163
 164   void remove_session(MonSession *s) {
 165     ceph_assert(!s->closed);
 166     for (std::map<std::string,Subscription*>::iterator p = s->sub_map.begin(); p != s->sub_map.end(); ++p) {
 167       p->second->type_item.remove_myself();
 168       delete p->second;
 169     }
 170     s->sub_map.clear();
 171     s->item.remove_myself();
 172     if (s->name.is_osd() &&
 173         s->name.num() >= 0) {
 174       for (auto p = by_osd.find(s->name.num());
 175            p->first == s->name.num();
 176            ++p)
 177         if (p->second == s) {
 178           by_osd.erase(p);
 179           break;
 180         }
 181     }
 182     if (s->con_features) {
 183       feature_map.rm(s->con_type, s->con_features);
 184     }
 185     s->closed = true;
 186     s->put();
 187   }
 188
 189   MonSession *new_session(const entity_name_t& n,
 190                           const entity_addrvec_t& av,
 191                           Connection *c) {
 192     MonSession *s = new MonSession(c);
 193     ceph_assert(s);
 194     s->_ident(n, av);
 195     add_session(s);
 196     return s;
 197   }
 198
 199   void add_session(MonSession *s) {
 200     s->session_timeout = ceph_clock_now();
 201     s->session_timeout += g_conf()->mon_session_timeout;
 202
 203     sessions.push_back(&s->item);
 204     s->get();
 205     if (s->name.is_osd() &&
 206         s->name.num() >= 0) {
 207       by_osd.insert(std::pair<int,MonSession*>(s->name.num(), s));
 208     }
 209     if (s->con_features) {
 210       feature_map.add(s->con_type, s->con_features);
 211     }
 212   }
 213
 214   MonSession *get_random_osd_session(OSDMap *osdmap) {
 215     // ok, this isn't actually random, but close enough.
 216     if (by_osd.empty())
 217       return 0;
 218     int n = by_osd.rbegin()->first + 1;
 219     int r = rand() % n;
 220
 221     auto p = by_osd.lower_bound(r);
 222     if (p == by_osd.end())
 223       --p;
 224
 225     if (!osdmap) {
 226       return p->second;
 227     }
 228
 229     MonSession *s = NULL;
 230
 231     auto b = p;
 232     auto f = p;
 233     bool backward = true, forward = true;
 234     while (backward || forward) {
 235       if (backward) {
 236         if (osdmap->is_up(b->first) &&
 237             osdmap->get_addrs(b->first) == b->second->con->get_peer_addrs()) {
 238           s = b->second;
 239           break;
 240         }
 241         if (b != by_osd.begin())
 242           --b;
 243         else
 244           backward = false;
 245       }
 246
 247       forward = (f != by_osd.end());
 248       if (forward) {
 249         if (osdmap->is_up(f->first)) {
 250           s = f->second;
 251           break;
 252         }
 253         ++f;
 254       }
 255     }
 256
 257     return s;
 258   }
 259
 260   void add_update_sub(MonSession *s, const std::string& what, version_t start, bool onetime, bool incremental_onetime) {
 261     Subscription *sub = 0;
 262     if (s->sub_map.count(what)) {
 263       sub = s->sub_map[what];
 264     } else {
 265       sub = new Subscription(s, what);
 266       s->sub_map[what] = sub;
 267
 268       if (!subs.count(what))
 269         subs[what] = new xlist<Subscription*>;
 270       subs[what]->push_back(&sub->type_item);
 271     }
 272     sub->next = start;
 273     sub->onetime = onetime;
 274     sub->incremental_onetime = onetime && incremental_onetime;
 275   }
 276
 277   void remove_sub(Subscription *sub) {
 278     sub->session->sub_map.erase(sub->type);
 279     sub->type_item.remove_myself();
 280     delete sub;
 281   }
 282 };
 283
 284 inline std::ostream& operator<<(std::ostream& out, const MonSession& s)
 285 {
 286   out << "MonSession(" << s.name << " " << s.addrs
 287       << " is " << (s.closed ? "closed" : "open")
 288       << " " << s.caps
 289       << ", features 0x" << std::hex << s.con_features << std::dec
 290       <<  " (" << ceph_release_name(ceph_release_from_features(s.con_features))
 291       << "))";
 292   return out;
 293 }
 294
 295 #endif